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