[Flink] Make Flink 2.x job submission work on REMOTE mode (#4500)

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

Six defects on the path that turns a saved Flink SQL application into a
submitted job. Each was found by submitting a real Flink SQL job to a
standalone cluster and fixing whatever failed next; they are independent
of each other but all sit on this one path.

1. ClassLoaderUtils.runAsClassLoader restored the context classloader
   captured in a static field when the class was first initialized —
   whichever thread happened to load it — instead of the one the calling
   thread had on entry. On pooled threads that silently replaces an
   unrelated thread's context classloader.

2. FlinkClientTrait.getCustomCommandLines and
   RemoteClient.getStandAloneClusterDescriptor call into Flink classes
   bound to the Flink version bundled with this module, while the calling
   thread's context classloader is FlinkShimsProxy's target-version shims
   classloader. Their internal ServiceLoader lookups therefore resolved
   providers from a different Flink version than the interfaces bundled
   here, failing with ServiceConfigurationError ("not a subtype"). Both
   call sites now run under their own class's classloader. Closes #4483.

3. The build-response getters (workspacePath, pass, shadedJarPath,
   flinkBaseImage, mainJarPath, extraLibJarPaths, flinkImageTag,
   podTemplatePaths, dockerInnerMainJarPath) do not follow JavaBean
   getter naming and carried no @JsonProperty, so Jackson silently
   skipped them: every build result persisted to t_flink_app's
   buildResultJson lost its paths, and only pass survived — by the
   coincidence that its field default is already true. A later submit
   then read back shadedJarPath == null and failed with an NPE, an
   "entry point class not found", or "flinkJobJar is null", depending on
   which downstream path consumed it.

4. SubmitRequest.userJarFile() passed shadedJarPath() straight to
   new File(...), which throws NPE when it is legitimately null.

5. streampark-console-service declared a compile dependency on
   streampark-flink-shims-base but not on streampark-flink-shims-base-v2,
   so FlinkTableInitializerV2 never reached the console's lib/ and every
   Flink 2.x SQL job failed with NoClassDefFoundError. Flink 1.x was
   unaffected, which is why this went unnoticed.

6. PackagedProgram's setUserClassPaths, disabled wholesale for #3761, is
   re-enabled for FLINK_SQL jobs only, so a SQL job's connector jars reach
   the client classpath. Verified against a real cluster not to reproduce
   the ClassCastException #3761 describes, and it leaves every other job
   type on the existing behaviour.

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

Submitting a Flink SQL job to a Flink 2.x cluster failed with a
NoClassDefFoundError long before reaching the cluster. Flink 1.x was
unaffected, which is why this went unnoticed. Five independent causes,
each of which only becomes visible after the previous one is fixed:

1. FlinkShimsProxy stopped putting the version-specific shims jar into
   the shims classloader. It matches on a name shaped
   "streampark-flink-shims_flink-<ver>_<scala>", which those artifacts
   carried until e770d2e renamed them without the Scala suffix. Both
   spellings are accepted now.

2. The same rename silenced the rule that pulls in the rest of
   StreamPark's Flink jars ("has a _<scala> suffix"), so the client stack
   was loaded by the console's own classloader and resolved
   org.apache.flink.* from the console's fixed baseline Flink instead of
   the target version's jars. That is the actual mechanism behind #4483:
   the ServiceLoader mismatch it reports is what a half-populated shims
   classloader looks like from the outside.

3. shims-base and shims-base-v2 share twelve class names — v2 redeclares
   them for Flink 2.x and inherits the rest — but both were added to
   every shims classloader regardless of the target version, in directory
   listing order. Which Flink version a class had been compiled for was
   therefore decided by the filesystem. A 1.x target no longer sees v2 at
   all, and a 2.x target gets v2 ahead of the base.

4. SavepointConfigOptions was removed in Flink 2.x, and Configuration's
   typed accessors (getBoolean/setBoolean/getInteger over a ConfigOption)
   went with it. Since this module is compiled once against a single
   baseline but submits to whichever version the user registered, both
   are now addressed portably: the savepoint options are declared from
   their keys, which are byte-identical across every supported version,
   and the generic get/set replace the typed accessors.

5. ClusterClient#submitJob widened its parameter from JobGraph to
   ExecutionPlan in 2.x. The instance satisfies either signature, only
   the declared type moved, so the call is made reflectively.

A FLINK_SQL program's classloader is also told to resolve
org.apache.streampark.* parent-first. The fat jar bundles whichever shims
it was built against, while the parent is the shims classloader for the
version actually registered; loading both ends in a LinkageError as soon
as one references the other.

Verified against real standalone clusters by driving StreamPark's own
submission path out-of-process: a Flink SQL job now submits and reaches
FINISHED on Flink 2.2.1, and the same job on Flink 1.20.4 — which worked
before this change — still does.

Not addressed: LocalClient and KubernetesNativeSessionClient use the same
removed Configuration accessors and will fail the same way on Flink 2.x.
Neither is reachable in the environment this was verified in, so they are
left for a change that can be tested.

Author: 88fantasy

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