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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
