Xiaobing Fang created FLINK-40740:
-------------------------------------
Summary: Fluss source can assign the same splits more than once
during periodic discovery
Key: FLINK-40740
URL: https://issues.apache.org/jira/browse/FLINK-40740
Project: Flink
Issue Type: Bug
Reporter: Xiaobing Fang
h2. Description
With periodic discovery enabled, \{{FlussSourceEnumerator}} can initialize the
same physical table path more than once before the first initialization
callback completes.
{\{checkTableBucketChanges}} excludes assigned and pending paths, but does not
track paths whose initialization is still in progress. A subsequent discovery
callback can therefore schedule another initialization for the same path.
When readers are registered, each initialization callback can assign the same
bucket splits again. Reassigning an earliest-offset split can rewind an active
reader and cause records to be emitted again.
Verified against master (3.7-SNAPSHOT), commit
\{{5c61d4ca327c05ed5964cfd8fc24727724123039}}.
h2. Steps to reproduce
# Register readers and discover a table with three buckets.
# Execute three discovery callbacks before processing any split-initialization
callback.
# Complete the queued initialization callbacks and inspect all split
assignments.
*Actual result:* Nine split assignments are produced for three buckets; each
bucket is assigned three times.
*Expected result:* Each bucket is initialized and assigned once during
discovery.
This interleaving is reproduced deterministically in an Enumerator regression
test. A full pipeline end-to-end reproduction was not run.
h2. Proposed fix
Track initializing physical table paths before submitting asynchronous split
initialization. Exclude these paths during subsequent discovery rounds, and
remove only the current request's paths when its callback finishes.
Keep this tracking local to the Enumerator without changing the checkpoint
format, so unfinished initialization can be rediscovered after recovery.
Preserve the existing initialization error propagation.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
