[
https://issues.apache.org/jira/browse/FLINK-40740?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40740:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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)
