From b28e7136b2fd5210a1e34b92c69ee58275bcfec5 Mon Sep 17 00:00:00 2001 From: 88fantasy <88fantasy@gmail.com> Date: Fri, 14 Aug 2026 10:18:49 +0800 Subject: [PATCH 1/2] [Flink] Fix Flink SQL job submission failures on the client path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../common/util/ClassLoaderUtils.java | 3 ++- .../streampark-console-service/pom.xml | 6 +++++ .../flink/client/bean/SubmitRequest.java | 3 ++- .../flink/client/impl/RemoteClient.java | 27 ++++++++++++++----- .../flink/client/trait/FlinkClientTrait.java | 27 ++++++++++++++++--- .../pipeline/AbstractFlinkBuildResponse.java | 2 ++ .../pipeline/DockerImageBuildResponse.java | 3 +++ .../pipeline/K8sAppModeBuildResponse.java | 3 +++ .../packer/pipeline/ShadedBuildResponse.java | 1 + 9 files changed, 63 insertions(+), 12 deletions(-) diff --git a/streampark-common/src/main/java/org/apache/streampark/common/util/ClassLoaderUtils.java b/streampark-common/src/main/java/org/apache/streampark/common/util/ClassLoaderUtils.java index 98080ab4b1..020568e107 100644 --- a/streampark-common/src/main/java/org/apache/streampark/common/util/ClassLoaderUtils.java +++ b/streampark-common/src/main/java/org/apache/streampark/common/util/ClassLoaderUtils.java @@ -39,11 +39,12 @@ private ClassLoaderUtils() { } public static R runAsClassLoader(ClassLoader targetClassLoader, Supplier 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); } } diff --git a/streampark-console/streampark-console-service/pom.xml b/streampark-console/streampark-console-service/pom.xml index b82cd03e55..dddc5ebaf3 100644 --- a/streampark-console/streampark-console-service/pom.xml +++ b/streampark-console/streampark-console-service/pom.xml @@ -372,6 +372,12 @@ ${project.version} + + org.apache.streampark + streampark-flink-shims-base-v2 + ${project.version} + + org.apache.streampark diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java index 350e9205c4..6128e5c73c 100644 --- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java +++ b/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; diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/impl/RemoteClient.java b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/impl/RemoteClient.java index 58e9024f91..7050dceda0 100644 --- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/impl/RemoteClient.java +++ b/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 executeClientAction( private Tuple2 getStandAloneClusterDescriptor( Configuration flinkConfig) { - DefaultClusterClientServiceLoader serviceLoader = new DefaultClusterClientServiceLoader(); - ClusterClientFactory 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 clientFactory = + serviceLoader.getClusterClientFactory(flinkConfig); + StandaloneClusterId standaloneClusterId = clientFactory.getClusterId(flinkConfig); + StandaloneClusterDescriptor standaloneClusterDescriptor = + (StandaloneClusterDescriptor) clientFactory.createClusterDescriptor(flinkConfig); + return new Tuple2<>(standaloneClusterId, standaloneClusterDescriptor); + }); } @FunctionalInterface diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java index ca5f66a3ce..7d495a5e65 100644 --- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java +++ b/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 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 getOptionFromDefaultFlinkConfig(String flinkHome, ConfigOption option) List 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) { diff --git a/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/AbstractFlinkBuildResponse.java b/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/AbstractFlinkBuildResponse.java index 2f80f3b4d9..d62b6ef4f6 100644 --- a/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/AbstractFlinkBuildResponse.java +++ b/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; } diff --git a/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/DockerImageBuildResponse.java b/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/DockerImageBuildResponse.java index 1d66f60a3c..76ce2f3615 100644 --- a/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/DockerImageBuildResponse.java +++ b/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 podTemplatePaths() { return podTemplatePaths; } + @JsonProperty("dockerInnerMainJarPath") public String dockerInnerMainJarPath() { return dockerInnerMainJarPath; } diff --git a/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/K8sAppModeBuildResponse.java b/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/K8sAppModeBuildResponse.java index 1124f3a9ab..9922044767 100644 --- a/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/K8sAppModeBuildResponse.java +++ b/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 extraLibJarPaths() { return extraLibJarPaths; } diff --git a/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/ShadedBuildResponse.java b/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/ShadedBuildResponse.java index f0370b54d8..e37af6fb82 100644 --- a/streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/ShadedBuildResponse.java +++ b/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; } From 8f6afd493adc6359202d90991d1aa2bd279516e3 Mon Sep 17 00:00:00 2001 From: 88fantasy <88fantasy@gmail.com> Date: Fri, 14 Aug 2026 11:24:13 +0800 Subject: [PATCH 2/2] [Flink] Make Flink 2.x job submission work on REMOTE mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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-_", which those artifacts carried until e770d2e8e 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 _ 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. --- .../flink/client/bean/SubmitRequest.java | 4 +- .../client/conf/FlinkSavepointOptions.java | 57 +++++++++++++++ .../client/tool/FlinkSessionSubmitHelper.java | 6 +- .../flink/client/trait/FlinkClientTrait.java | 71 ++++++++++++++----- .../flink/proxy/FlinkShimsProxy.java | 53 ++++++++++++-- 5 files changed, 164 insertions(+), 27 deletions(-) create mode 100644 streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/conf/FlinkSavepointOptions.java diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java index 6128e5c73c..be1d7941ec 100644 --- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java +++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java @@ -30,6 +30,7 @@ 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; @@ -37,7 +38,6 @@ 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 { diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/conf/FlinkSavepointOptions.java b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/conf/FlinkSavepointOptions.java new file mode 100644 index 0000000000..ee5ba6f4ed --- /dev/null +++ b/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. + * + *

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. + * + *

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 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 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() { + } +} diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/tool/FlinkSessionSubmitHelper.java b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/tool/FlinkSessionSubmitHelper.java index 364f9e6a54..e2b24a3f39 100644 --- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/tool/FlinkSessionSubmitHelper.java +++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/tool/FlinkSessionSubmitHelper.java @@ -20,6 +20,7 @@ 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; @@ -27,7 +28,6 @@ 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 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() { diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java index 7d495a5e65..a9bc73c988 100644 --- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java +++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java @@ -40,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; @@ -70,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; @@ -222,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, @@ -236,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(), @@ -386,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, @@ -418,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 @@ -509,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. + * + *

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 getJobGraph( Configuration flinkConfig, SubmitRequest submitRequest, File jarFile) throws Exception { @@ -535,16 +571,18 @@ public Tuple2 getJobGraph( } else { builder.setJarFile(jarFile); 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. + // 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()); } } @@ -617,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) { diff --git a/streampark-flink/streampark-flink-proxy/src/main/java/org/apache/streampark/flink/proxy/FlinkShimsProxy.java b/streampark-flink/streampark-flink-proxy/src/main/java/org/apache/streampark/flink/proxy/FlinkShimsProxy.java index 4a1d60d856..d4f9fd242e 100644 --- a/streampark-flink/streampark-flink-proxy/src/main/java/org/apache/streampark/flink/proxy/FlinkShimsProxy.java +++ b/streampark-flink/streampark-flink-proxy/src/main/java/org/apache/streampark/flink/proxy/FlinkShimsProxy.java @@ -35,6 +35,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Comparator; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -62,6 +63,14 @@ public final class FlinkShimsProxy extends LoggerSupport { private static final String FLINK_SHIMS_BASE_PREFIX = "streampark-flink-shims-base"; + private static final String FLINK_SHIMS_BASE_V2_PREFIX = "streampark-flink-shims-base-v2"; + + private static final String STREAMPARK_PREFIX = "streampark-"; + + private static final String STREAMPARK_CONSOLE_PREFIX = "streampark-console"; + + private static final String STREAMPARK_SHADED_PREFIX = "streampark-shaded-"; + private static final List PARENT_FIRST_PATTERNS = Collections.unmodifiableList( Arrays.asList( "java.", @@ -79,6 +88,13 @@ public final class FlinkShimsProxy extends LoggerSupport { private FlinkShimsProxy() { } + /** {@code majorVersion} is shaped like "1.20" or "2.2". */ + private static boolean isFlink2(String majorVersion) { + int dot = majorVersion.indexOf('.'); + String major = dot < 0 ? majorVersion : majorVersion.substring(0, dot); + return "2".equals(major); + } + private static Pattern getFlinkShimsResourcePattern(String majorVersion) { return Pattern.compile( "flink-(.*)-" + majorVersion + "(.*).jar", @@ -180,6 +196,7 @@ private static void addShimsUrls(FlinkVersion flinkVersion, Consumer addSh return; } + List matched = new ArrayList<>(); for (File jar : jars) { String jarName = jar.getName(); if (!jarName.endsWith(Constants.JAR_SUFFIX)) { @@ -187,25 +204,51 @@ private static void addShimsUrls(FlinkVersion flinkVersion, Consumer addSh } String includeReason = matchShimIncludeReason(jarName, majorVersion, scalaVersion); if (includeReason != null) { - addShimUrl.accept(jar); + matched.add(jar); LOG.logInfo(includeReason + jarName); } } + // shims-base-v2 must precede shims-base: it redeclares a subset of its classes for Flink + // 2.x, and a URL classloader takes the first match. Directory listing order is arbitrary, + // so leaving this to chance means the Flink version a class was compiled for is decided by + // the filesystem. + matched.sort( + Comparator.comparing(file -> file.getName().startsWith(FLINK_SHIMS_BASE_V2_PREFIX) ? 0 : 1)); + matched.forEach(addShimUrl); } private static String matchShimIncludeReason( String jarName, String majorVersion, String scalaVersion) { if (jarName.startsWith(FLINK_SHIMS_PREFIX)) { - String prefixVer = FLINK_SHIMS_PREFIX + "-" + majorVersion + "_" + scalaVersion; - return jarName.startsWith(prefixVer) ? "Include flink shims jar lib: " : null; + // The shims artifacts carried a _${scala.binary.version} suffix until they were renamed + // without it; both spellings are accepted so the jar is matched either way. Getting this + // wrong is silent — a non-matching jar is simply left out of the classloader, and the + // job fails much later with a NoClassDefFoundError for a class the shims needed. + String prefixVer = FLINK_SHIMS_PREFIX + "-" + majorVersion; + return jarName.startsWith(prefixVer + "_" + scalaVersion) || jarName.startsWith(prefixVer + "-") + ? "Include flink shims jar lib: " + : null; } if (jarName.startsWith(FLINK_SHIMS_BASE_PREFIX)) { - return "Include flink shims base jar lib: "; + // shims-base-v2 redeclares a subset of shims-base for Flink 2.x and inherits the rest, + // so a 2.x target needs both jars with v2 taking precedence (see addShimsUrls), while a + // 1.x target must not see v2 at all — the two share class names, and v2's copies are + // compiled against APIs that moved in 2.x. + return !jarName.startsWith(FLINK_SHIMS_BASE_V2_PREFIX) || isFlink2(majorVersion) + ? "Include flink shims base jar lib: " + : null; } if (INCLUDE_PATTERN.matcher(jarName).matches()) { return "Include jar lib: "; } - if (jarName.matches("^streampark-.*_" + scalaVersion + ".*$")) { + // Everything StreamPark builds against Flink belongs in here, so that the whole submission + // stack resolves org.apache.flink.* from the target version's jars rather than from the + // console's own bundled baseline. This used to be spelled as "has a _${scala.binary.version} + // suffix", which stopped selecting anything on the Flink side once those modules were + // renamed without the suffix. The console's own jars stay out: they are the parent. + if (jarName.startsWith(STREAMPARK_PREFIX) + && !jarName.startsWith(STREAMPARK_CONSOLE_PREFIX) + && !jarName.startsWith(STREAMPARK_SHADED_PREFIX)) { return "Include streampark lib: "; } return null;