Xiaobing Fang created FLINK-40973:
-------------------------------------
Summary: 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
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)
