[FLINK-40040][cdc-runtime] Wrap coordinator failure in SerializedThrowable across the RPC boundary#4456
Open
Zhile wants to merge 4 commits into
Open
[FLINK-40040][cdc-runtime] Wrap coordinator failure in SerializedThrowable across the RPC boundary#4456Zhile wants to merge 4 commits into
Zhile wants to merge 4 commits into
Conversation
Member
|
Could you please check the failing test cases? |
Author
|
@yuxiqian please help review. I also pasted the full exception in the description. Please refer to that. |
Member
|
Could you please add a pipeline E2e case for this patch? I believe we can reproduce this issue by putting different jar dependencies in JM and TM docker containers. Also, rebasing latest |
added 3 commits
July 8, 2026 11:24
…wable across RPC boundary SchemaRegistry.failJob now wraps the failure cause into a SerializedThrowable before it crosses the operator-coordinator RPC boundary, and runInEventLoop routes through failJob so all exits share the wrapping. Otherwise an exception whose class only lives in the user classloader (e.g. the MySQL driver's com.mysql.cj.exceptions.ConnectionIsClosedException raised during table discovery) fails to deserialize on the receiving side, where flink-rpc-akka uses an isolated classloader, with ClassNotFoundException. The coordination response is then lost and the SchemaOperator request stalls until rpcTimeout, failing with a misleading TimeoutException and turning a transient error into a restart loop that only a full job restart recovers from.
…lizedThrowable-wrapped coordinator failures Since the coordinator now wraps its failure into a SerializedThrowable before completing the pending coordination response (so it survives the operator-coordinator RPC boundary), the exact typed exception chain is no longer observable by the requester. Update testExceptionEvolveSchema and testEvolveSchemaWithFailure to assert on stack trace content instead: - The top-level exception type is not asserted because the failure may surface via the operator request path or the job-failure path depending on timing. - Only strings present in all printed representations are asserted: the wrapping FlinkRuntimeException message and the original exception's class name (the wrapped cause is printed either as its fully-qualified class name or via the exception's custom toString, depending on the delivery path).
…lizedThrowable-wrapped coordinator failures Since the coordinator failure now crosses the operator-coordinator RPC boundary as a SerializedThrowable, the "Caused by: " rendering of the cause chain in the JobManager log varies with the delivery path. Drop the "Caused by: " prefixes from the expected log lines and only wait for the class names / messages that appear in every representation. The tolerant-mode expectations are unchanged, since those WARN logs print the original exception before it is wrapped.
…th JobManager-only exception class
Reproduces the production classpath asymmetry: a jar containing
JobManagerOnlyException is installed into the JobManager container's
/opt/flink/lib only, never shipped to the TaskManager container or
submitted with the pipeline jars. The values sink gains an
error.on.schema.change.exception.class option so its metadata applier
raises that class (instantiated reflectively on the coordinator) as the
cause of a rejected schema change.
Verified against the pre-fix runtime: without the SerializedThrowable
wrapping in SchemaRegistry.failJob, the exceptionally completed
coordination response cannot be deserialized by the TaskManager's RPC
layer, which tears down the JobManager <-> TaskManager Pekko association
("Association with remote system ... has failed, address is now gated.
Reason: [org.apache.flink.cdc...Exception]"). Cancellation then stalls
for a full RPC timeout cycle before the job reaches FAILED, and the
original cause never surfaces on the TaskManager. The test asserts the
job fails cleanly and that the TaskManager's RPC layer never chokes on
the coordinator failure; it fails on the pre-fix runtime and passes
after the fix (19s vs 79s runtime).
Author
|
please help review. |
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
2 participants
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
Schema-evolution coordination responses are dropped when the failure's cause is a class that only lives in the user classloader (e.g.
com.mysql.cj.exceptions.ConnectionIsClosedException).flink-rpc-akkadeserializes the coordination response with an isolated classloader and throwsClassNotFoundException, so the response never reachesSchemaOperator, which then blocks onresponseFuture.get(rpcTimeout)and fails with a misleadingTimeoutException-- turning a transient error into a restart loop that only a full job restart recovers from.Observed in production (Flink CDC 3.6.0 on Flink 1.20.3):
3 minutes later (exactly the schema operator rpcTimeout), the pending request times out:
JIRA: https://issues.apache.org/jira/browse/FLINK-40040
Brief change log
SchemaRegistry.failJobnow wraps the cause intoSerializedThrowable(idempotent, not re-wrapped if already one) before callinghandleUnrecoverableError(which completes pending coordination responses exceptionally) andcontext.failJob.SerializedThrowablecarries the original exception as bytes plus a stringified stack trace, so the receiving side can deserialize it without the original class and still see the real cause.SchemaRegistry.runInEventLoop's catch block now routes throughfailJobinstead of duplicating the fail path, so every exit shares the wrapping.SchemaCoordinator, since both complete pending response futures via the sharedfailJob->handleUnrecoverableErrorpath.Verifying this change
Added
SchemaRegistryFailJobSerializationTest:failJobWrapsExceptionIntoSerializedThrowableasserts the failure delivered to the coordinator context is aSerializedThrowable.wrappedFailureSurvivesIsolatedClassloaderButRawOneDoesNotshows a raw exception whose cause is a classloader-isolated type fails to deserialize (ClassNotFoundException), while theSerializedThrowable-wrapped one deserializes and preserves the real cause text.Does this pull request potentially affect one of the following parts