fix(flink): close lookup reader after cache reload attempts - #19503
fix(flink): close lookup reader after cache reload attempts#19503danny0405 wants to merge 1 commit into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR tightens up resource cleanup for the Flink lookup reader by closing the table reader in a finally on every cache-reload attempt and making open()/close() idempotent with proper rollback and suppressed-exception handling. The rollback ordering and idempotency logic look correct; one minor point about exception masking in the reload finally is worth a look in the inline comment. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here.
| RowData rowData = serializer.copy(row); | ||
| RowData key = extractLookupKey(rowData); | ||
| cache.addRow(key, rowData); | ||
| } |
There was a problem hiding this comment.
🤖 With the new finally { partitionReader.close(); }, if the read loop throws and close() also throws (which it now can, since it rethrows format/rich close failures), the finally's exception replaces the original read failure — so the outer catch logs/wraps the close exception instead of the true root cause. Given the care taken elsewhere in this PR to preserve causes via addSuppressed, would it be worth doing the same here (e.g. try-with-resources or catch-and-suppress) so the reload's real failure isn't masked in the retry log at line 180?
There was a problem hiding this comment.
Addressed in 536fed5. HoodieLookupTableReader now implements Closeable, and each reload attempt uses try-with-resources. If reading fails and cleanup also fails, the read failure remains primary and the cleanup failure is attached as suppressed instead of masking it.
| IOException exception = null; | ||
| try { | ||
| format.close(); | ||
| } catch (IOException e) { |
There was a problem hiding this comment.
format.close() can throw an unchecked HoodieIOException in the production COW path: ParquetSplitRecordIterator.close() converts its reader's IOException into that runtime exception. Since this catch handles only IOException, such a failure skips closeInputFormat(); it can also replace the original open() failure during rollback because the catch there likewise suppresses only checked close failures. Please preserve both checked and unchecked cleanup failures while still attempting the rich-format cleanup, and add a regression test using a runtime split-close failure.
There was a problem hiding this comment.
Addressed in 536fed5. open() now uses Hudi’s closeSuppressing helper during rollback, and close() uses try-with-resources to guarantee closeInputFormat() runs after both checked and runtime split-close failures while preserving suppression order. Added regression tests with HoodieIOException for runtime rollback and runtime split-close failures. The 7 focused lookup tests pass.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR makes the Flink lookup table reader Closeable and closes it on every cache-reload attempt (including failures) via try-with-resources, with idempotent cleanup and suppressed-exception preservation on the open/rollback and close paths. Tracing the open/rollback, close ordering, retry-loop reuse, and the production input-format close() implementations, the resource-cleanup logic looks correct and the earlier-round concerns appear addressed. No new issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review. One small naming nit in the close() cleanup path; everything else looks clean.
cc @yihua
| this.inputSplits = null; | ||
| if (format == null) { | ||
| return; | ||
| } |
There was a problem hiding this comment.
🤖 nit: ignored reads like the variable is inconsequential, but it's actually the vehicle that guarantees closeInputFormat is called even when format.close() throws. Something like formatCloser would signal that the Closeable itself is the mechanism, not an accident.
4 participants
Describe the issue this Pull Request addresses
Flink lookup cache reloads closed the table reader only after a complete, successful read. If opening or reading an input format failed, the retry opened another input format while the previous attempt could still hold file handles or native reader resources.
Summary and Changelog
No code was copied from another project.
Impact
No public API, configuration, storage-format, or performance behavior changes. Failed Flink lookup cache reloads no longer accumulate input-format resources across local retries.
Risk Level
low. The change is limited to lookup reader resource cleanup and is covered by focused success and failure-path tests.
Documentation Update
none.
Contributor's checklist
Testing
mvn -pl hudi-flink-datasource/hudi-flink -am -DskipITs -Dcheckstyle.skip -Drat.skip=true -Dtest=TestHoodieLookupFunction,TestHoodieLookupTableReader -Dsurefire.failIfNoSpecifiedTests=false test(7 tests passed)mvn -pl hudi-flink-datasource/hudi-flink checkstyle:check -Dcheckstyle.includes='**/table/lookup/*.java'(0 violations)