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