From 6969986fdfa04ac4361352c4d41466e0332b5c94 Mon Sep 17 00:00:00 2001
From: Piotr Przybylski <piotrprz@gmail.com>
Date: Sat, 25 Apr 2026 00:56:08 +0200
Subject: [PATCH 1/3] Test with Scala

---
 .../flink-end-to-end-tests-table-api/pom.xml               | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-table-api/pom.xml b/flink-end-to-end-tests/flink-end-to-end-tests-table-api/pom.xml
index be819aa3c7af1..d7e88114adf44 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-table-api/pom.xml
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-table-api/pom.xml
@@ -39,6 +39,13 @@ under the License.
 			<scope>provided</scope>
 		</dependency>
 
+		<!-- Incompatible Scala library in user code shouldn't break table-planner -->
+		<dependency>
+			<groupId>org.scala-lang</groupId>
+			<artifactId>scala-library</artifactId>
+			<version>2.13.18</version>
+		</dependency>
+
 		<!-- Flink Table API -->
 		<dependency>
 			<groupId>org.apache.flink</groupId>

From 52321912466b7019d10e90cb2794e97be8957e69 Mon Sep 17 00:00:00 2001
From: Piotr Przybylski <piotrprz@gmail.com>
Date: Sat, 25 Apr 2026 17:40:25 +0200
Subject: [PATCH 2/3] Add failing test

---
 .../join/JoinWithCustomTypeExampleTest.java   | 29 +++++++++++++++++++
 1 file changed, 29 insertions(+)
 create mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-table-api/src/test/java/org/apache/flink/table/test/join/JoinWithCustomTypeExampleTest.java

diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-table-api/src/test/java/org/apache/flink/table/test/join/JoinWithCustomTypeExampleTest.java b/flink-end-to-end-tests/flink-end-to-end-tests-table-api/src/test/java/org/apache/flink/table/test/join/JoinWithCustomTypeExampleTest.java
new file mode 100644
index 0000000000000..0e51c0dbc5f95
--- /dev/null
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-table-api/src/test/java/org/apache/flink/table/test/join/JoinWithCustomTypeExampleTest.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.test.join;
+
+import org.junit.jupiter.api.Test;
+
+public class JoinWithCustomTypeExampleTest {
+
+    @Test
+    void testJoinWithCustomTypeExample() throws Exception {
+        JoinWithCustomTypeExample.main(new String[0]);
+    }
+}

From 368459d973da235911a91bab680f22387c098e1a Mon Sep 17 00:00:00 2001
From: Piotr Przybylski <piotrprz@gmail.com>
Date: Sat, 25 Apr 2026 20:16:13 +0200
Subject: [PATCH 3/3] Add fix

---
 .../operators/join/adaptive/AdaptiveJoinOperatorFactory.java   | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/adaptive/AdaptiveJoinOperatorFactory.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/adaptive/AdaptiveJoinOperatorFactory.java
index 7246f3718c07f..890060082ae0b 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/adaptive/AdaptiveJoinOperatorFactory.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/adaptive/AdaptiveJoinOperatorFactory.java
@@ -102,9 +102,10 @@ public StreamOperatorFactory<?> genOperatorFactory(
         ClassLoader classLoader =
                 plannerModule == null
                         ? userClassLoader
-                        : FlinkUserCodeClassLoaders.parentFirst(
+                        : FlinkUserCodeClassLoaders.childFirst(
                                 plannerModule.getSubmoduleClassLoader().getURLs(),
                                 userClassLoader,
+                                new String[0],
                                 NOOP_EXCEPTION_HANDLER,
                                 checkClassLoaderLeak);