[Flink] Fix Flink SQL job submission failures on the client path

streampark-common/src/main/java/org/apache/streampark/common/util/ClassLoaderUtils.java

@@ -39,11 +39,12 @@ private ClassLoaderUtils() {
     }
 
     public static <R> R runAsClassLoader(ClassLoader targetClassLoader, Supplier<R> supplier) {
+        ClassLoader previousClassLoader = Thread.currentThread().getContextClassLoader();
         try {
             Thread.currentThread().setContextClassLoader(targetClassLoader);
             return supplier.get();
         } finally {
-            Thread.currentThread().setContextClassLoader(ORIGINAL_CLASS_LOADER);
+            Thread.currentThread().setContextClassLoader(previousClassLoader);
         }
     }
 

streampark-console/streampark-console-service/pom.xml

@@ -372,6 +372,12 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.streampark</groupId>
+            <artifactId>streampark-flink-shims-base-v2</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <!-- Ensure all Flink shims are built when console-service is built with -am -->
         <dependency>
             <groupId>org.apache.streampark</groupId>

streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java

@@ -296,7 +296,8 @@ public File userJarFile() {
             } else {
                 checkBuildResult();
                 ShadedBuildResponse shadedBuildResult = buildResult.as(ShadedBuildResponse.class);
-                userJarFile = new File(shadedBuildResult.shadedJarPath());
+                String shadedJarPath = shadedBuildResult.shadedJarPath();
+                userJarFile = shadedJarPath == null ? null : new File(shadedJarPath);
             }
         }
         return userJarFile;

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/impl/RemoteClient.java

@@ -17,6 +17,7 @@
 
 package org.apache.streampark.flink.client.impl;
 
+import org.apache.streampark.common.util.ClassLoaderUtils;
 import org.apache.streampark.flink.client.bean.CancelRequest;
 import org.apache.streampark.flink.client.bean.CancelResponse;
 import org.apache.streampark.flink.client.bean.SavepointRequestTrait;
@@ -165,13 +166,25 @@ private <O, R extends SavepointRequestTrait> O executeClientAction(
 
     private Tuple2<StandaloneClusterId, StandaloneClusterDescriptor> getStandAloneClusterDescriptor(
                                                                                                     Configuration flinkConfig) {
-        DefaultClusterClientServiceLoader serviceLoader = new DefaultClusterClientServiceLoader();
-        ClusterClientFactory<StandaloneClusterId> clientFactory =
-            serviceLoader.getClusterClientFactory(flinkConfig);
-        StandaloneClusterId standaloneClusterId = clientFactory.getClusterId(flinkConfig);
-        StandaloneClusterDescriptor standaloneClusterDescriptor =
-            (StandaloneClusterDescriptor) clientFactory.createClusterDescriptor(flinkConfig);
-        return new Tuple2<>(standaloneClusterId, standaloneClusterDescriptor);
+        // DefaultClusterClientServiceLoader is bound to the Flink version bundled with this module
+        // (loaded by this class's own classloader), but the calling thread's context classloader may
+        // currently be a target-version shims classloader (see FlinkShimsProxy). Its internal
+        // ServiceLoader.load(ClusterClientFactory.class) resolves providers via the context
+        // classloader, so leaving it as the shims classloader here would load a ClusterClientFactory
+        // implementation from a different Flink version than the interface bundled here, throwing
+        // ServiceConfigurationError ("not a subtype"). Force it back to this class's own classloader
+        // for the duration of this call.
+        return ClassLoaderUtils.runAsClassLoader(
+            RemoteClient.class.getClassLoader(),
+            () -> {
+                DefaultClusterClientServiceLoader serviceLoader = new DefaultClusterClientServiceLoader();
+                ClusterClientFactory<StandaloneClusterId> clientFactory =
+                    serviceLoader.getClusterClientFactory(flinkConfig);
+                StandaloneClusterId standaloneClusterId = clientFactory.getClusterId(flinkConfig);
+                StandaloneClusterDescriptor standaloneClusterDescriptor =
+                    (StandaloneClusterDescriptor) clientFactory.createClusterDescriptor(flinkConfig);
+                return new Tuple2<>(standaloneClusterId, standaloneClusterDescriptor);
+            });
     }
 
     @FunctionalInterface

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java

@@ -26,6 +26,7 @@
 import org.apache.streampark.common.enums.FlinkRestoreMode;
 import org.apache.streampark.common.fs.FsOperator;
 import org.apache.streampark.common.util.AssertUtils;
+import org.apache.streampark.common.util.ClassLoaderUtils;
 import org.apache.streampark.common.util.DeflaterUtils;
 import org.apache.streampark.common.util.ExceptionUtils;
 import org.apache.streampark.common.util.FlinkConfigurationUtils;
@@ -533,8 +534,18 @@ public Tuple2<PackagedProgram, JobGraph> getJobGraph(
             }
         } else {
             builder.setJarFile(jarFile);
-            // BUG: https://github.com/apache/streampark/issues/3761
-            // .setUserClassPaths(Lists.newArrayList(submitRequest.classPaths()))
+            if (submitRequest.jobType() == FlinkJobType.FLINK_SQL) {
+                // FLINK_SQL fat jar only bundles the SQL client + shims; it does not contain the
+                // target Flink version's own connector/runtime jars, so those must be added
+                // explicitly. Scoped to FLINK_SQL only, unlike the blanket disable from
+                // https://github.com/apache/streampark/issues/3761, which was never verified
+                // against this job type on REMOTE mode specifically. Note this does not help
+                // resolve org.apache.flink.* classes themselves: Flink's classloader always
+                // resolves that package parent-first, and PackagedProgram's parent is
+                // console's own bundled (baseline-version) flink-clients, not the target
+                // version, so those still fail when target and baseline Flink versions diverge.
+                builder.setUserClassPaths(Lists.newArrayList(submitRequest.classPaths()));
+            }
         }
 
         PackagedProgram packageProgram = builder.build();
@@ -587,7 +598,17 @@ <T> T getOptionFromDefaultFlinkConfig(String flinkHome, ConfigOption<T> option)
     List<CustomCommandLine> getCustomCommandLines(String flinkHome) {
         Configuration flinkDefaultConfiguration = getFlinkDefaultConfiguration(flinkHome);
         String confDir = flinkHome + "/conf";
-        return CliFrontend.loadCustomCommandLines(flinkDefaultConfiguration, confDir);
+        // CliFrontend/GenericCLI are bound to the Flink version bundled with this module (loaded by
+        // this class's own classloader), but the calling thread's context classloader may currently
+        // be a target-version shims classloader (see FlinkShimsProxy). GenericCLI's internal
+        // ServiceLoader.load(PipelineExecutorFactory.class) resolves providers via the context
+        // classloader, so leaving it as the shims classloader here would load a PipelineExecutorFactory
+        // implementation from a different Flink version than the interface bundled here, throwing
+        // ServiceConfigurationError ("not a subtype"). Force it back to this class's own classloader
+        // for the duration of this call.
+        return ClassLoaderUtils.runAsClassLoader(
+            FlinkClientTrait.class.getClassLoader(),
+            () -> CliFrontend.loadCustomCommandLines(flinkDefaultConfiguration, confDir));
     }
 
     public Integer getParallelism(SubmitRequest submitRequest) {

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/AbstractFlinkBuildResponse.java

@@ -40,11 +40,13 @@ protected AbstractFlinkBuildResponse(String workspacePath, boolean pass) {
     }
 
     @Override
+    @JsonProperty("workspacePath")
     public String workspacePath() {
         return workspacePath;
     }
 
     @Override
+    @JsonProperty("pass")
     public boolean pass() {
         return pass;
     }

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/DockerImageBuildResponse.java

@@ -52,14 +52,17 @@ public DockerImageBuildResponse(
         this.dockerInnerMainJarPath = dockerInnerMainJarPath;
     }
 
+    @JsonProperty("flinkImageTag")
     public String flinkImageTag() {
         return flinkImageTag;
     }
 
+    @JsonProperty("podTemplatePaths")
     public Map<String, String> podTemplatePaths() {
         return podTemplatePaths;
     }
 
+    @JsonProperty("dockerInnerMainJarPath")
     public String dockerInnerMainJarPath() {
         return dockerInnerMainJarPath;
     }

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/K8sAppModeBuildResponse.java

@@ -44,14 +44,17 @@ public K8sAppModeBuildResponse(
         this.extraLibJarPaths = extraLibJarPaths;
     }
 
+    @JsonProperty("flinkBaseImage")
     public String flinkBaseImage() {
         return flinkBaseImage;
     }
 
+    @JsonProperty("mainJarPath")
     public String mainJarPath() {
         return mainJarPath;
     }
 
+    @JsonProperty("extraLibJarPaths")
     public Set<String> extraLibJarPaths() {
         return extraLibJarPaths;
     }

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/ShadedBuildResponse.java

@@ -37,6 +37,7 @@ public ShadedBuildResponse(String workspacePath, String shadedJarPath, boolean p
     public ShadedBuildResponse() {
     }
 
+    @JsonProperty("shadedJarPath")
     public String shadedJarPath() {
         return shadedJarPath;
     }