[Flink] Make Flink 2.x job submission work on REMOTE mode

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

@@ -30,14 +30,14 @@
 import org.apache.streampark.common.util.DeflaterUtils;
 import org.apache.streampark.common.util.HdfsUtils;
 import org.apache.streampark.common.util.PropertiesUtils;
+import org.apache.streampark.flink.client.conf.FlinkSavepointOptions;
 import org.apache.streampark.flink.packer.pipeline.BuildResult;
 import org.apache.streampark.flink.packer.pipeline.ShadedBuildResponse;
 
 import org.apache.streampark.shaded.com.fasterxml.jackson.core.type.TypeReference;
 import org.apache.streampark.shaded.com.fasterxml.jackson.databind.ObjectMapper;
 
 import org.apache.commons.collections.MapUtils;
-import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 
 import javax.annotation.Nullable;
@@ -263,7 +263,7 @@ public String flinkSQL() {
     public boolean allowNonRestoredState() {
         if (allowNonRestoredState == null) {
             Object value =
-                properties.get(SavepointConfigOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE.key());
+                properties.get(FlinkSavepointOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE.key());
             if (value == null) {
                 allowNonRestoredState = false;
             } else {
@@ -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-api/src/main/java/org/apache/streampark/flink/client/conf/FlinkSavepointOptions.java

@@ -0,0 +1,57 @@
+/*
+ * 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.streampark.flink.client.conf;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/**
+ * The savepoint restore options, declared here rather than taken from Flink.
+ *
+ * <p>Flink kept moving the class that declares them — {@code
+ * org.apache.flink.runtime.jobgraph.SavepointConfigOptions} through 1.x, {@code
+ * org.apache.flink.configuration.StateRecoveryOptions} in 2.x — while keeping the option *keys*
+ * byte-identical across every version this project supports. Since this module is compiled once
+ * against a single baseline Flink but submits to whichever version the user registered, referring
+ * to either class binds the client to one version family and fails against the other with a
+ * {@code NoClassDefFoundError} at submission time. Declaring the options from their keys sidesteps
+ * that entirely: a {@code Configuration} is keyed by string, so a locally declared option addresses
+ * exactly the same setting as Flink's own.
+ *
+ * <p>The keys are part of Flink's public configuration surface, so they are as stable as the user's
+ * own {@code flink-conf.yaml} entries.
+ */
+public final class FlinkSavepointOptions {
+
+    /** Mirrors Flink's {@code execution.savepoint.path}. */
+    public static final ConfigOption<String> SAVEPOINT_PATH =
+        ConfigOptions.key("execution.savepoint.path")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Path to a savepoint to restore the job from.");
+
+    /** Mirrors Flink's {@code execution.savepoint.ignore-unclaimed-state}. */
+    public static final ConfigOption<Boolean> SAVEPOINT_IGNORE_UNCLAIMED_STATE =
+        ConfigOptions.key("execution.savepoint.ignore-unclaimed-state")
+            .booleanType()
+            .defaultValue(false)
+            .withDescription("Allow to skip savepoint state that cannot be restored.");
+
+    private FlinkSavepointOptions() {
+    }
+}

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/tool/FlinkSessionSubmitHelper.java

@@ -20,14 +20,14 @@
 import org.apache.streampark.common.util.AssertUtils;
 import org.apache.streampark.common.util.JsonUtils;
 import org.apache.streampark.common.util.LoggerSupport;
+import org.apache.streampark.flink.client.conf.FlinkSavepointOptions;
 import org.apache.streampark.flink.kubernetes.KubernetesRetriever;
 
 import org.apache.streampark.shaded.com.fasterxml.jackson.databind.JsonNode;
 
 import org.apache.flink.client.deployment.application.ApplicationConfiguration;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
-import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
 import org.apache.hc.client5.http.entity.mime.MultipartEntityBuilder;
 import org.apache.hc.client5.http.fluent.Request;
 import org.apache.hc.core5.http.ContentType;
@@ -161,9 +161,9 @@ class JarRunRequest {
         List<String> args = flinkConf.get(ApplicationConfiguration.APPLICATION_ARGS);
         this.programArgs = args == null ? null : String.join(" ", args);
         this.parallelism = String.valueOf(flinkConf.get(CoreOptions.DEFAULT_PARALLELISM));
-        this.savepointPath = flinkConf.get(SavepointConfigOptions.SAVEPOINT_PATH);
+        this.savepointPath = flinkConf.get(FlinkSavepointOptions.SAVEPOINT_PATH);
         this.allowNonRestoredState =
-            flinkConf.getBoolean(SavepointConfigOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE);
+            flinkConf.get(FlinkSavepointOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE);
     }
 
     public String getEntryClass() {

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;
@@ -39,6 +40,7 @@
 import org.apache.streampark.flink.client.bean.SubmitRequest;
 import org.apache.streampark.flink.client.bean.SubmitResponse;
 import org.apache.streampark.flink.client.bean.TriggerSavepointRequest;
+import org.apache.streampark.flink.client.conf.FlinkSavepointOptions;
 import org.apache.streampark.flink.core.FlinkClusterClient;
 import org.apache.streampark.flink.core.conf.FlinkRunOption;
 
@@ -69,7 +71,6 @@
 import org.apache.flink.configuration.PipelineOptionsInternal;
 import org.apache.flink.python.PythonOptions;
 import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.Preconditions;
 
@@ -221,6 +222,27 @@ protected void logSavepointClientRequest(String operation, SavepointRequestTrait
         logInfo(message.toString());
     }
 
+    /**
+     * Submits a job graph, tolerating the signature change {@code ClusterClient#submitJob} went
+     * through: it took a {@code JobGraph} until Flink 2.x widened the parameter to {@code
+     * ExecutionPlan}, which {@code JobGraph} implements. The instance is accepted by either
+     * version — only the declared parameter type moved — but this module is compiled once against
+     * a single baseline, so a direct call binds to one signature and fails against the other with
+     * {@code NoSuchMethodError}.
+     */
+    private static String submitJobGraph(ClusterClient<?> client, JobGraph jobGraph) throws Exception {
+        for (java.lang.reflect.Method method : client.getClass().getMethods()) {
+            if ("submitJob".equals(method.getName())
+                && method.getParameterCount() == 1
+                && method.getParameterTypes()[0].isInstance(jobGraph)) {
+                Object future = method.invoke(client, jobGraph);
+                return ((java.util.concurrent.CompletableFuture<?>) future).get().toString();
+            }
+        }
+        throw new FlinkException(
+            "No ClusterClient#submitJob(..) accepting a JobGraph on " + client.getClass().getName());
+    }
+
     protected SubmitResponse submitJobGraphToCluster(
                                                      SubmitRequest submitRequest,
                                                      Configuration flinkConfig,
@@ -235,7 +257,7 @@ protected SubmitResponse submitJobGraphToCluster(
                 PackagedProgram packageProgram = programJobGraph._1();
                 JobGraph jobGraph = programJobGraph._2();
                 ClusterClient<?> client = clientSupplier.call();
-                String jobId = client.submitJob(jobGraph).get().toString();
+                String jobId = submitJobGraph(client, jobGraph);
                 SubmitResponse result =
                     new SubmitResponse(
                         clusterIdSupplier.call(),
@@ -385,7 +407,7 @@ private void applyPyFlinkConfig(SubmitRequest submitRequest, Configuration flink
     private void applyCommonPipelineConfig(SubmitRequest submitRequest, Configuration flinkConfig) {
         safeSet(flinkConfig, PipelineOptions.NAME, submitRequest.effectiveAppName());
         safeSet(flinkConfig, DeploymentOptions.TARGET, submitRequest.deployMode().getName());
-        safeSet(flinkConfig, SavepointConfigOptions.SAVEPOINT_PATH, submitRequest.savePoint());
+        safeSet(flinkConfig, FlinkSavepointOptions.SAVEPOINT_PATH, submitRequest.savePoint());
         safeSet(
             flinkConfig,
             ApplicationConfiguration.APPLICATION_MAIN_CLASS,
@@ -417,10 +439,10 @@ private void applySavepointConfig(SubmitRequest submitRequest, Configuration fli
         }
         safeSet(
             flinkConfig,
-            SavepointConfigOptions.SAVEPOINT_PATH,
+            FlinkSavepointOptions.SAVEPOINT_PATH,
             submitRequest.savePoint());
-        flinkConfig.setBoolean(
-            SavepointConfigOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE,
+        flinkConfig.set(
+            FlinkSavepointOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE,
             submitRequest.allowNonRestoredState());
         boolean enableRestoreMode =
             submitRequest.restoreMode() != null
@@ -508,6 +530,21 @@ protected SubmitResponse trySubmit(
         }
     }
 
+    /**
+     * Makes the program's classloader resolve {@code org.apache.streampark.*} from its parent — the
+     * shims classloader built for the registered Flink version — instead of from the submitted jar.
+     *
+     * <p>Set through the raw key rather than {@code CoreOptions}: the typed accessors around it
+     * moved between Flink 1.x and 2.x, while the key itself did not, and this module is compiled
+     * against a single baseline but submits to whichever version the user registered.
+     */
+    private static Configuration streamParkParentFirstConfig() {
+        Configuration configuration = new Configuration();
+        configuration.setString(
+            "classloader.parent-first-patterns.additional", "org.apache.streampark.");
+        return configuration;
+    }
+
     public Tuple2<PackagedProgram, JobGraph> getJobGraph(
                                                          Configuration flinkConfig, SubmitRequest submitRequest,
                                                          File jarFile) throws Exception {
@@ -533,8 +570,20 @@ 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) {
+                // The FLINK_SQL fat jar bundles only the SQL client and the shims it was built
+                // against; it carries none of the target Flink version's own jars, so those have to
+                // be handed to the program explicitly. Scoped to FLINK_SQL, unlike the blanket
+                // disable from https://github.com/apache/streampark/issues/3761, which was never
+                // verified against this job type.
+                builder.setUserClassPaths(Lists.newArrayList(submitRequest.classPaths()));
+                // ...and the StreamPark classes must come from the parent — this thread runs under
+                // the shims classloader for the *registered* Flink version, whereas the fat jar
+                // carries whichever shims it happened to be built with. Loading both ends in a
+                // LinkageError as soon as one references the other, and silently mixes Flink
+                // versions when it does not.
+                builder.setConfiguration(streamParkParentFirstConfig());
+            }
         }
 
         PackagedProgram packageProgram = builder.build();
@@ -587,7 +636,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) {
@@ -596,8 +655,7 @@ public Integer getParallelism(SubmitRequest submitRequest) {
                 submitRequest.getProp(ConfigKeys.KEY_FLINK_PARALLELISM()).toString());
         }
         return getFlinkDefaultConfiguration(submitRequest.flinkVersion().getFlinkHome())
-            .getInteger(
-                CoreOptions.DEFAULT_PARALLELISM, CoreOptions.DEFAULT_PARALLELISM.defaultValue());
+            .get(CoreOptions.DEFAULT_PARALLELISM, CoreOptions.DEFAULT_PARALLELISM.defaultValue());
     }
 
     Options getCommandLineOptions(String flinkHome) {

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;
     }