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..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 { @@ -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-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/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/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 ca5f66a3ce..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 @@ -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. + * + *

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 { @@ -533,8 +570,20 @@ 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) { + // 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 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) { @@ -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) { 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; } 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;