cshuo opened a new pull request, #19315:
URL: https://github.com/apache/hudi/pull/19315
### Describe the issue this Pull Request addresses
Closes #19314.
Flink can remove an idle split fetcher from its fetcher map and create its
replacement before the old fetcher finishes shutting down. `HoodieSourceReader`
previously supplied a new `HoodieSourceSplitReader` that captured the same
stateful `SplitReaderFunction` instance for every fetcher generation. The old
fetcher's asynchronous `close()` could therefore clear the iterator being used
by the new fetcher.
Synchronizing iterator access is insufficient because the old `close()` may
still run between the new fetcher's `open()` and `readBatch()` calls. The
stateful reader function needs exclusive ownership instead.
### Summary and Changelog
- Pass a serializable `SplitReaderFunction` supplier through
`HoodieTableSource`, `HoodieSource`, and `HoodieSourceReader`.
- Instantiate a fresh reader function for every
`HoodieSourceSplitReader`/fetcher generation.
- Apply the lifecycle isolation to both regular and CDC split reader
functions.
- Add a regression test proving that closing one split reader does not close
the reader function owned by another split reader.
No code was copied.
### Impact
This prevents an idle fetcher's delayed shutdown from asynchronously closing
the active Flink Source V2 reader cursor. There are no configuration,
storage-format, or user-facing API changes.
### Risk Level
Low. The change is limited to reader-function construction and ownership.
Existing split-reader tests and the new lifecycle regression test pass.
Validation:
```bash
mvn -pl hudi-flink-datasource/hudi-flink -am \
-Dtest=TestHoodieSourceSplitReader \
-Dsurefire.failIfNoSpecifiedTests=false \
-DskipITs -DskipSparkTests -DskipScalaTests \
test
```
Tests run: 28, failures: 0, errors: 0, skipped: 0.
### Documentation Update
None.
### Contributor's checklist
- [ ] Read through [contributor's
guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] 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]
