[Flink] Make Flink 2.x job submission work on REMOTE mode - #4500
Open
88fantasy wants to merge 2 commits into
Open
[Flink] Make Flink 2.x job submission work on REMOTE mode#450088fantasy wants to merge 2 commits into
88fantasy wants to merge 2 commits into
Conversation
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 apache#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 apache#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 apache#3761 describes, and it leaves every other job
type on the existing behaviour.
Submitting a Flink SQL job to a Flink 2.x cluster failed with a NoClassDefFoundError long before reaching the cluster. Flink 1.x was unaffected, which is why this went unnoticed. Five independent causes, each of which only becomes visible after the previous one is fixed: 1. FlinkShimsProxy stopped putting the version-specific shims jar into the shims classloader. It matches on a name shaped "streampark-flink-shims_flink-<ver>_<scala>", which those artifacts carried until e770d2e renamed them without the Scala suffix. Both spellings are accepted now. 2. The same rename silenced the rule that pulls in the rest of StreamPark's Flink jars ("has a _<scala> suffix"), so the client stack was loaded by the console's own classloader and resolved org.apache.flink.* from the console's fixed baseline Flink instead of the target version's jars. That is the actual mechanism behind apache#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.
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
Makes Flink 2.x job submission work on REMOTE mode. Today no Flink 2.x SQL job can be submitted at all, while Flink 1.x is unaffected — which is why this went unnoticed.
Closes #4499
Brief change log
Five independent causes on one path — each only becomes visible once the previous is fixed:
FlinkShimsProxywas not loading the per-version shims jar. It matches a name shapedstreampark-flink-shims_flink-<major>_<scala>, which those artifacts carried untile770d2e8erenamed them without the Scala suffix. Both spellings are accepted now.The same rename silenced the rule that pulls in the rest of StreamPark's Flink jars (
^streampark-.*_<scala>.*$), so the whole client stack was loaded by the console's own classloader and resolvedorg.apache.flink.*from the console's fixed baseline Flink rather than the registered target version. That, I believe, is the actual mechanism behind [Bug] PipelineExecutorFactory ServiceConfigurationError when submitting a Flink job (ChildFirstClassLoader / console baseline Flink version conflict) #4483 — theServiceLoadermismatch it reports is what a half-populated shims classloader looks like from the outside.shims-baseandshims-base-v2share twelve class names (v2 redeclares them for Flink 2.x, inheriting the rest) yet both were added to every shims classloader inlistFiles()order — so which Flink version a class was compiled for was decided by the filesystem. A 1.x target no longer sees v2 at all; a 2.x target gets v2 ahead of the base.API removed in Flink 2.x.
SavepointConfigOptionsis gone (2.x declares the same keys inStateRecoveryOptions), andConfiguration's typed accessors over aConfigOptionwent with it. Since this module is compiled once against a single baseline but submits to whichever version the user registered, both are addressed portably: the savepoint options are declared from their keys — byte-identical across 1.17–2.3, verified against the distributions — and the genericget/setreplace the typed accessors.ClusterClient#submitJobwidened its parameter fromJobGraphtoExecutionPlanin 2.x. The instance satisfies either signature, so the call is made reflectively.Additionally, a
FLINK_SQLprogram's classloader is told to resolveorg.apache.streampark.*parent-first: the fat jar bundles whichever shims it was built against, while the parent is now the shims classloader for the version actually registered, and loading both ends in aLinkageErroras soon as one references the other.Verifying this change
Verified against real standalone clusters by driving StreamPark's own submission path out-of-process against the deployed console's
lib/, so each fix could be confirmed against the actual artifacts:FINISHEDon Flink 2.2.1 — previously impossible.org.apache.flink.util.ParameterToolresolves fromflink-dist-2.2.1.jar; without them it fails exactly as reported.Then confirmed through the console itself, not just the extracted path: a deployed build submits a
Flink SQL job (mysql-cdc -> Paimon, two sinks) to the 2.2.1 cluster, it reaches
FINISHED, and thejob appears downstream exactly as a Flink 1.x job does.
Does this pull request potentially affect one of the following parts
Touches
FlinkShimsProxy, whichAGENTS.mdmarks high-sensitivity. No static state is introduced; the changes are to which jars go into the classloader and in what order.Not addressed:
LocalClientandKubernetesNativeSessionClientuse the same removedConfigurationaccessors and will fail the same way on Flink 2.x. Neither is reachable in the environment this was verified in, so I have left them for a change that can be tested rather than guessed at.Documentation
e770d2e8e) actually work.