[I] [Flink] Source V2 reuses a stateful SplitReaderFunction across fetcher generations [hudi]

["cshuo (via GitHub)" <gi...@apache.org>]

cshuo opened a new issue, #19314:
URL: https://github.com/apache/hudi/issues/19314

   ### Problem
   
   `HoodieSourceReader` creates a new `HoodieSourceSplitReader` for each fetcher, but its supplier captures and reuses the same stateful `SplitReaderFunction` instance:
   
   `hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieSourceReader.java`
   
   This allows two consecutive split fetchers to access the same reader function during the fetcher handoff.
   
   ### Race condition
   
   The race can occur as follows:
   
   1. Flink detects that a fetcher is idle, calls `shutdown(true)`, and immediately removes it from the fetcher map. The old fetcher may not have completed its final shutdown yet.
   2. A new split arrives. Because the fetcher map is empty, `SingleThreadFetcherManager.addSplits()` creates and starts a new fetcher with a new `HoodieSourceSplitReader`.
   3. The old fetcher exits on its own thread and calls:
      `SplitReader.close() -> HoodieSourceSplitReader.close() -> SplitReaderFunction.close()`.
   4. Meanwhile, the new fetcher may already be using that same `SplitReaderFunction` instance in `readBatch()`.
   5. The old fetcher's asynchronous `close()` clears state belonging to the new fetcher, such as the cursor's `currentIterator`.
   
   The close delegation is in:
   
   `hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieSourceSplitReader.java`
   
   This may cause the active fetcher to observe an unexpectedly cleared iterator and fail or stop reading prematurely.
   
   ### Root cause
   
   Consecutive `HoodieSourceSplitReader` instances share one stateful `SplitReaderFunction`. Fetcher generations have overlapping lifetimes during idle shutdown, so ownership of that function is not exclusive.
   
   ### Why synchronization is insufficient
   
   Synchronizing `hasNext()`, `next()`, and `close()` can prevent the iterator from being cleared between `hasNext()` and `next()`, but it does not eliminate the lifecycle race. The old fetcher's `close()` can still run after the new fetcher has called `open()` and before its next `readBatch()`.
   
   ### Expected fix
   
   Each `HoodieSourceSplitReader` should own an independent `SplitReaderFunction` instance. Pass a serializable factory/supplier through the source construction path and instantiate a fresh function whenever Flink creates a split reader.
   
   The fix should cover both regular and CDC split reader functions and include a regression test proving that closing one split reader does not close or mutate the function owned by another split reader.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] Source V2 reuses a stateful SplitReaderFunction across fetcher generations [hudi]

["danny0405 (via GitHub)" <gi...@apache.org>]

danny0405 closed issue #19314: Source V2 reuses a stateful SplitReaderFunction across fetcher generations
URL: https://github.com/apache/hudi/issues/19314


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org