Skip to content

[FLINK-39150][runtime] Fix join operator crashes jobs when using custom types or custom type serializers#27697

Open
noorall wants to merge 3 commits intoapache:masterfrom
noorall:fix-39150
Open

[FLINK-39150][runtime] Fix join operator crashes jobs when using custom types or custom type serializers#27697
noorall wants to merge 3 commits intoapache:masterfrom
noorall:fix-39150

Conversation

@noorall
Copy link
Contributor

@noorall noorall commented Feb 27, 2026

What is the purpose of the change

The PlannerComponentClassLoader uses a strict whitelist-based routing strategy and does not automatically fall back to the parent ClassLoader for packages that are not whitelisted.

  • Whitelisted packages (e.g., org.apache.flink.*): follow the configured lookup order (e.g., parent-first/component-first) and can fall back accordingly.
  • Non-whitelisted packages (e.g., custom user POJOs): default to a component-only lookup. Since these classes exist only in the user JAR (parent) and not in the isolated table-planner-*.jar (component), deserialization fails with ClassNotFoundException.

Required fix:

  1. Change the default loading behavior for non-whitelisted packages so that for non-whitelisted packages, it will automatically fall back to the parent/component ClassLoader. This ensures custom user types can be resolved during the AdaptiveJoin deserialization.
  2. Use UserClassLoader instead of the current thread context ClassLoader.

Brief change log

  • Fix join operator crashes jobs when using custom types or custom type serializers

Does this pull request potentially affect one of the following parts:

  • Dependencies (does it add or upgrade a dependency): (yes / no)
  • The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
  • The serializers: (yes / no / don't know)
  • The runtime per-record code paths (performance sensitive): (yes / no / don't know)
  • Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
  • The S3 file system connector: (yes / no / don't know)

Documentation

  • Does this pull request introduce a new feature? (yes / no)
  • If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

@flinkbot
Copy link
Collaborator

flinkbot commented Feb 27, 2026

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

Copy link
Contributor

@zhuzhurk zhuzhurk left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM.
Thanks for looking into this and getting it fixed! @noorall

} catch (ClassNotFoundException | IOException e) {
return false;
throw new RuntimeException(
"Failed to deserialize AdaptiveJoin instance. "
Copy link
Contributor

@davidradl davidradl Feb 27, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When would the flink-table-planner-loader.jar not be in the class path.
Some thoughts:

  • I would think this situation issue is more about not being able to loa a class that the deserialization requires. Have I misunderstood?
  • the class not found exception seems to occur when a Class of a serialized object cannot be found in the readObject on the stream. Are expecting this to be in the jar.
  • do all the io exceptions also imply that the flink-table-planner-loader.jar not be in the class path.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Typically this situation should not happen. The existing workflow assumes that the current thread context ClassLoader or the table planner ClassLoader can resolve all required classes (which is a wrong assumption). Therefore, once this exception occurs, the most common cause is a missing flink-table-planner-loader.jar. The message is mainly meant to be user-friendly and to aid troubleshooting, and it does not imply this is the only possible root cause.

  1. Your understanding is correct.
  2. The missing classes are not necessarily all contained in the planner-related jars; they can also come from the user jar.
  3. As explained above, a missing flink-table-planner-loader.jar is the most common scenario, so this log message should be treated as a suggestive troubleshooting hint rather than a definitive diagnosis.

@piotrp
Copy link
Contributor

piotrp commented Mar 2, 2026

Looks like this test I provided verifies only MiniCluster behavior, where user classes are mixed with Flink classes. Even after applying your patch Flink's JobManager still crashes after throwing an exception after job JAR is uploaded:

2026-03-02 01:13:16.471 | 2026-03-02 00:13:16,469 ERROR org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Fatal error occurred in the cluster entrypoint.
2026-03-02 01:13:16.471 | org.apache.flink.util.FlinkException: JobMaster for job 73e1c9595b191dae276e39958e99435a failed.
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.dispatcher.Dispatcher.jobMasterFailed(Dispatcher.java:1942) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.dispatcher.Dispatcher.jobManagerRunnerFailed(Dispatcher.java:1010) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.dispatcher.Dispatcher.lambda$runJob$12(Dispatcher.java:932) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.CompletableFuture$Completion.run(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.MdcUtils.lambda$wrapRunnable$1(MdcUtils.java:70) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRunAsync$4(PekkoRpcActor.java:460) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRunAsync(PekkoRpcActor.java:460) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:225) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.FencedPekkoRpcActor.handleRpcMessage(FencedPekkoRpcActor.java:88) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleMessage(PekkoRpcActor.java:174) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:37) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:33) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction.applyOrElse(PartialFunction.scala:127) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:33) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.Actor.aroundReceive(Actor.scala:547) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.Actor.aroundReceive$(Actor.scala:545) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.AbstractActor.aroundReceive(AbstractActor.scala:229) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:273) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:234) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.dispatch.Mailbox.exec(Mailbox.scala:246) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | Caused by: org.apache.flink.runtime.jobmaster.JobMasterException: Could not update the state of task execution for JobMaster.
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:544) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	... 23 more
2026-03-02 01:13:16.471 | Caused by: java.lang.RuntimeException: java.lang.RuntimeException: Failed to deserialize AdaptiveJoin instance. Please check whether the flink-table-planner-loader.jar is in the classpath.
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.handleJobEvent(DefaultAdaptiveExecutionHandler.java:86) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.lambda$notifyJobVertexFinishedIfPossible$17(AdaptiveBatchScheduler.java:660) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.Optional.ifPresent(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.notifyJobVertexFinishedIfPossible(AdaptiveBatchScheduler.java:658) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.onTaskFinished(AdaptiveBatchScheduler.java:406) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:833) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:813) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:534) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	... 23 more
2026-03-02 01:13:16.471 | Caused by: java.lang.RuntimeException: Failed to deserialize AdaptiveJoin instance. Please check whether the flink-table-planner-loader.jar is in the classpath.
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.lazyInitialize(AdaptiveJoinOperatorFactory.java:134) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.checkAndLazyInitialize(AdaptiveJoinOperatorFactory.java:96) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.getJoinType(AdaptiveJoinOperatorFactory.java:78) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.AdaptiveBroadcastJoinOptimizationStrategy.tryOptimizeAdaptiveJoin(AdaptiveBroadcastJoinOptimizationStrategy.java:105) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.BaseAdaptiveJoinOperatorOptimizationStrategy.visitDownstreamAdaptiveJoinNode(BaseAdaptiveJoinOperatorOptimizationStrategy.java:90) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.AdaptiveBroadcastJoinOptimizationStrategy.onOperatorsFinished(AdaptiveBroadcastJoinOptimizationStrategy.java:74) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.StreamGraphOptimizer.onOperatorsFinished(StreamGraphOptimizer.java:72) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.tryOptimizeStreamGraph(DefaultAdaptiveExecutionHandler.java:118) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.handleJobEvent(DefaultAdaptiveExecutionHandler.java:82) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.lambda$notifyJobVertexFinishedIfPossible$17(AdaptiveBatchScheduler.java:660) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.Optional.ifPresent(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.notifyJobVertexFinishedIfPossible(AdaptiveBatchScheduler.java:658) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.onTaskFinished(AdaptiveBatchScheduler.java:406) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:833) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:813) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:534) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	... 23 more
2026-03-02 01:13:16.471 | Caused by: java.lang.ClassNotFoundException: org.apache.notflink.TableJoinTestClass
2026-03-02 01:13:16.471 | 	at java.base/java.net.URLClassLoader.findClass(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:197) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.Class.forName0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.Class.forName(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:76) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readNonProxyDesc(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readClassDesc(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readClass(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.ArrayList.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.GeneratedMethodAccessor13.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectStreamClass.invokeReadObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:488) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:472) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:467) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.lazyInitialize(AdaptiveJoinOperatorFactory.java:132) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.checkAndLazyInitialize(AdaptiveJoinOperatorFactory.java:96) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.getJoinType(AdaptiveJoinOperatorFactory.java:78) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.AdaptiveBroadcastJoinOptimizationStrategy.tryOptimizeAdaptiveJoin(AdaptiveBroadcastJoinOptimizationStrategy.java:105) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.BaseAdaptiveJoinOperatorOptimizationStrategy.visitDownstreamAdaptiveJoinNode(BaseAdaptiveJoinOperatorOptimizationStrategy.java:90) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.AdaptiveBroadcastJoinOptimizationStrategy.onOperatorsFinished(AdaptiveBroadcastJoinOptimizationStrategy.java:74) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.StreamGraphOptimizer.onOperatorsFinished(StreamGraphOptimizer.java:72) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.tryOptimizeStreamGraph(DefaultAdaptiveExecutionHandler.java:118) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.handleJobEvent(DefaultAdaptiveExecutionHandler.java:82) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.lambda$notifyJobVertexFinishedIfPossible$17(AdaptiveBatchScheduler.java:660) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.Optional.ifPresent(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.notifyJobVertexFinishedIfPossible(AdaptiveBatchScheduler.java:658) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.onTaskFinished(AdaptiveBatchScheduler.java:406) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:833) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:813) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:534) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	... 23 more

@noorall
Copy link
Contributor Author

noorall commented Mar 2, 2026

Looks like this test I provided verifies only MiniCluster behavior, where user classes are mixed with Flink classes. Even after applying your patch Flink's JobManager still crashes after throwing an exception after job JAR is uploaded:

2026-03-02 01:13:16.471 | 2026-03-02 00:13:16,469 ERROR org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Fatal error occurred in the cluster entrypoint.
2026-03-02 01:13:16.471 | org.apache.flink.util.FlinkException: JobMaster for job 73e1c9595b191dae276e39958e99435a failed.
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.dispatcher.Dispatcher.jobMasterFailed(Dispatcher.java:1942) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.dispatcher.Dispatcher.jobManagerRunnerFailed(Dispatcher.java:1010) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.dispatcher.Dispatcher.lambda$runJob$12(Dispatcher.java:932) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.CompletableFuture$Completion.run(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.MdcUtils.lambda$wrapRunnable$1(MdcUtils.java:70) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRunAsync$4(PekkoRpcActor.java:460) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRunAsync(PekkoRpcActor.java:460) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:225) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.FencedPekkoRpcActor.handleRpcMessage(FencedPekkoRpcActor.java:88) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleMessage(PekkoRpcActor.java:174) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:37) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:33) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction.applyOrElse(PartialFunction.scala:127) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:33) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.Actor.aroundReceive(Actor.scala:547) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.Actor.aroundReceive$(Actor.scala:545) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.AbstractActor.aroundReceive(AbstractActor.scala:229) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:273) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:234) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.pekko.dispatch.Mailbox.exec(Mailbox.scala:246) [flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source) [?:?]
2026-03-02 01:13:16.471 | Caused by: org.apache.flink.runtime.jobmaster.JobMasterException: Could not update the state of task execution for JobMaster.
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:544) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	... 23 more
2026-03-02 01:13:16.471 | Caused by: java.lang.RuntimeException: java.lang.RuntimeException: Failed to deserialize AdaptiveJoin instance. Please check whether the flink-table-planner-loader.jar is in the classpath.
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.handleJobEvent(DefaultAdaptiveExecutionHandler.java:86) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.lambda$notifyJobVertexFinishedIfPossible$17(AdaptiveBatchScheduler.java:660) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.Optional.ifPresent(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.notifyJobVertexFinishedIfPossible(AdaptiveBatchScheduler.java:658) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.onTaskFinished(AdaptiveBatchScheduler.java:406) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:833) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:813) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:534) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	... 23 more
2026-03-02 01:13:16.471 | Caused by: java.lang.RuntimeException: Failed to deserialize AdaptiveJoin instance. Please check whether the flink-table-planner-loader.jar is in the classpath.
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.lazyInitialize(AdaptiveJoinOperatorFactory.java:134) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.checkAndLazyInitialize(AdaptiveJoinOperatorFactory.java:96) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.getJoinType(AdaptiveJoinOperatorFactory.java:78) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.AdaptiveBroadcastJoinOptimizationStrategy.tryOptimizeAdaptiveJoin(AdaptiveBroadcastJoinOptimizationStrategy.java:105) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.BaseAdaptiveJoinOperatorOptimizationStrategy.visitDownstreamAdaptiveJoinNode(BaseAdaptiveJoinOperatorOptimizationStrategy.java:90) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.AdaptiveBroadcastJoinOptimizationStrategy.onOperatorsFinished(AdaptiveBroadcastJoinOptimizationStrategy.java:74) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.StreamGraphOptimizer.onOperatorsFinished(StreamGraphOptimizer.java:72) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.tryOptimizeStreamGraph(DefaultAdaptiveExecutionHandler.java:118) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.handleJobEvent(DefaultAdaptiveExecutionHandler.java:82) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.lambda$notifyJobVertexFinishedIfPossible$17(AdaptiveBatchScheduler.java:660) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.Optional.ifPresent(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.notifyJobVertexFinishedIfPossible(AdaptiveBatchScheduler.java:658) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.onTaskFinished(AdaptiveBatchScheduler.java:406) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:833) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:813) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:534) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	... 23 more
2026-03-02 01:13:16.471 | Caused by: java.lang.ClassNotFoundException: org.apache.notflink.TableJoinTestClass
2026-03-02 01:13:16.471 | 	at java.base/java.net.URLClassLoader.findClass(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:197) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.Class.forName0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.Class.forName(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:76) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readNonProxyDesc(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readClassDesc(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readClass(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.util.ArrayList.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.GeneratedMethodAccessor13.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectStreamClass.invokeReadObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream$FieldValues.<init>(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:488) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:472) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:467) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.lazyInitialize(AdaptiveJoinOperatorFactory.java:132) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.checkAndLazyInitialize(AdaptiveJoinOperatorFactory.java:96) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoinOperatorFactory.getJoinType(AdaptiveJoinOperatorFactory.java:78) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.AdaptiveBroadcastJoinOptimizationStrategy.tryOptimizeAdaptiveJoin(AdaptiveBroadcastJoinOptimizationStrategy.java:105) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.BaseAdaptiveJoinOperatorOptimizationStrategy.visitDownstreamAdaptiveJoinNode(BaseAdaptiveJoinOperatorOptimizationStrategy.java:90) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.table.runtime.strategy.AdaptiveBroadcastJoinOptimizationStrategy.onOperatorsFinished(AdaptiveBroadcastJoinOptimizationStrategy.java:74) ~[flink-table-runtime-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.StreamGraphOptimizer.onOperatorsFinished(StreamGraphOptimizer.java:72) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.tryOptimizeStreamGraph(DefaultAdaptiveExecutionHandler.java:118) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.DefaultAdaptiveExecutionHandler.handleJobEvent(DefaultAdaptiveExecutionHandler.java:82) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.lambda$notifyJobVertexFinishedIfPossible$17(AdaptiveBatchScheduler.java:660) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/java.util.Optional.ifPresent(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.notifyJobVertexFinishedIfPossible(AdaptiveBatchScheduler.java:658) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchScheduler.onTaskFinished(AdaptiveBatchScheduler.java:406) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:833) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:813) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:534) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[flink-dist-2.3-SNAPSHOT.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229) ~[flink-rpc-akkae6f85170-a441-4a28-b81e-ab5fdfbeb1ca.jar:2.3-SNAPSHOT]
2026-03-02 01:13:16.471 | 	... 23 more

I’ve completed the fix and verified it locally. Could you please try again on your side?

…hen using custom types or custom type serializers
@piotrp
Copy link
Contributor

piotrp commented Mar 2, 2026

Thanks, test job passed 👍

Would it be possible to add a proper regression test for this, i.e. one that actually tests with a user JAR?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

5 participants