wombatu-kun opened a new pull request, #19202: URL: https://github.com/apache/hudi/pull/19202
### Describe the issue this Pull Request addresses `ITTestHoodieDataSource.testStreamReadFromSpecifiedCommitWithChangelog` is flaky on the Flink CDC streaming-read path. It intermittently fails with `AssertionError: Unexpected job failure` whose real cause is a `NullPointerException` in `CdcIterators$BaseImageIterator.hasNext`, driven by `CdcFileSplitsIterator.hasNext` and `BatchRecords.nextRecordFromSplit`. It was observed on the CI of an unrelated PR: https://github.com/apache/hudi/actions/runs/28791442071/job/85370887969 (shard `test-flink-1 (flink2.1)`; every other Flink shard passed). This path only recently started being exercised by default after native CDC read landed (#19114) and the default log format switched to native (#19118), which is why the flake surfaces now. ### Summary and Changelog Root cause is a cross-thread data race reachable only on the read teardown path. `CdcFileSplitsIterator` is drained on the Flink task thread (lazily, via `BatchRecords.nextRecordFromSplit`), but Flink can `close()` it from the split-fetcher thread during job teardown. Its `recordIterator` / `imageManager` fields (and the nested `BaseImageIterator.cdcItr`) are non-volatile: read on the task thread and nulled by `close()` on the fetcher thread. When a streaming read is force-terminated mid-drain (the test uses a `CollectSinkTableFactory.SuccessException` to stop the job once the expected rows are collected), `close()` races the in-flight read and the task thread dereferences a nulled field, producing the NPE. In a real job this is confined to cancel/finish/failover and is masked by the restart strategy, but the data race itself is in production code. - `CdcIterators.CdcFileSplitsIterator`: make `hasNext()`, `next()`, and `close()` `synchronized` on `this`, so a concurrent `close()` can no longer null out iterator state mid-read. This single lock also covers every nested per-split iterator and the `imageManager`, since those are only ever touched through this instance. Add a `closed` flag so that after `close()` the iterator reports end-of-stream (`hasNext()` returns `false`, `next()` throws `NoSuchElementException`) instead of touching already-released state, and make `close()` idempotent. In steady state the monitor is uncontended (only the task thread drains), so the read-path cost is negligible; the only contention is at teardown, where `close()` now correctly waits for an in-flight read. - `ITTestHoodieDataSource`: because `BatchRecords` calls `hasNext()` then `next()` as two separate calls, a teardown `close()` landing between them still ends a force-terminated drain with a benign `NoSuchElementException` from the now-closed iterator; with `restart-strategy.fixed-delay.attempts=0` (set in `beforeEach` for determinism) that becomes the job's reported failure even though the sink already collected its rows. Extend the existing `isAcceptableTerminalFailure` teardown-race tolerance (previously scoped to the `ParquetColumnarRowSplitReader#readNextRowGroup` frame) with a fourth cause: an NPE or `NoSuchElementException` originating from a `CdcIterators` nested-iterator frame. `submitAndFetchWithRetry` already retries idempotent short reads, so a race firing before all rows are collected is still covered. ### Impact No user-facing or public-API change. The production change hardens the Flink CDC read iterator against a teardown-time data race that could otherwise surface as a spurious NPE during job cancel/finish/failover. The test change removes a source of CI flakiness on the CDC streaming-read ITs. ### Risk Level low The production change is confined to the CDC read iterator's lifecycle methods: it preserves the existing single-threaded iteration semantics and only adds mutual exclusion with `close()` plus a well-defined post-close end-of-stream. Verified by building the `flink2.1` / Java 11 configuration that matches the failing shard and running `ITTestHoodieDataSource#testStreamReadFromSpecifiedCommitWithChangelog` locally. The race is non-deterministic, so correctness rests primarily on the happens-before reasoning above; the test-side tolerance is a safety net for the forced-teardown termination that no production change can remove. ### Documentation Update none ### Contributor's checklist - [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [ ] Enough context is provided in the sections above - [ ] Adequate tests were added if applicable -- 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]
