[
https://issues.apache.org/jira/browse/FLINK-40973?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40973:
-----------------------------------
Labels: pull-request-available (was: )
> Fluss source fails on retriable errors during periodic table discovery
> ----------------------------------------------------------------------
>
> Key: FLINK-40973
> URL: https://issues.apache.org/jira/browse/FLINK-40973
> Project: Flink
> Issue Type: Improvement
> Components: Flink CDC
> Reporter: Xiaobing Fang
> Priority: Not a Priority
> Labels: pull-request-available
>
> h2. Description
> With periodic discovery enabled, FlussSourceEnumerator fails the job when a
> discovery call encounters a transient Fluss metadata error.
> For example, listPartitionInfos().get() may throw an ExecutionException
> wrapping a Fluss NetworkException during a coordinator leader change.
> checkTableBucketChanges() currently converts every discovery error into:
> {code}
> FlinkRuntimeException: Failed to discover subscribed table-buckets.
> {code}
> The exception escapes the coordinator callback and triggers job failover.
> With restarts disabled, the job terminates instead of waiting for the next
> discovery cycle.
> h2. Steps to reproduce
> The error-handling path is reproduced deterministically through the
> Enumerator's asynchronous discovery callback:
> # Enable periodic discovery and complete a successful discovery round.
> # Make the next TableDiscoverer.discover() call throw an ExecutionException
> wrapping org.apache.fluss.exception.NetworkException.
> # Execute the periodic discovery callback.
> Actual result: The callback throws FlinkRuntimeException, although another
> discovery cycle is scheduled.
> Expected result: Retain the last successful subscription and assignment
> state, then retry on the next discovery cycle.
> Reproduced against master commit 68ae5222c (3.8-SNAPSHOT). The regression
> injects the wrapped exception through TableDiscoverer; it does not simulate a
> real coordinator leader failure.
> h2. Proposed fix
> * When scanDiscoveryIntervalMs > 0 and the exception cause chain contains
> org.apache.fluss.exception.RetriableException, log a warning and return
> before applying any discovery result.
> * Preserve subscriptions, assigned and pending splits, and pending-removal
> state. Do not convert a failed discovery into a successful empty result.
> * Continue to fail for non-retriable errors and one-time discovery. Leave
> split-initialization error handling unchanged.
> This refines the discovery failure policy described in FLINK-40575 while
> preserving the distinction between a successful empty subscription and a
> failed discovery.
> h2. Verification
> The three regression instances that require tolerating the transient failure
> fail before the fix and pass afterward. Coverage includes initial discovery,
> existing assignments, pending assignments, pending removals, subsequent
> recovery, and errors that must still fail.
> * Java 11 / Flink 1.20.3: 9 focused tests passed.
> * Java 17 / Flink 2.2.0: 6 focused tests passed.
> * No failures, errors, or skipped tests in either successful run. Spotless
> and Checkstyle passed.
> A full pipeline leader-failover end-to-end test was not run.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
