Skip to content

fix(flink): add close function in AbstractStreamWriteFunction #18743

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
HuangZhenQiu wants to merge 1 commit into from
+8 −0
Closed

fix(flink): add close function in AbstractStreamWriteFunction #18743

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions ...rce/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,14 @@ public void endInput() {
this.inputEnded = true;
}

@Override
public void close() throws Exception {
if (this.writeClient != null) {
Copy link
Copy Markdown
Contributor

@hudi-agent hudi-agent May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Should this also call super.close() to chain to the parent (ProcessFunctionAdapter / RichFunction)? Subclasses like StreamWriteFunction and IndexWriteFunction invoke super.close() at the end of their own close(), which now lands here — and stops, leaving anything the Flink base class needs to clean up unhandled. Wrapping the writeClient.close() in a try/finally with super.close() in the finally would also ensure cleanup proceeds if the write client throws.

- AI-generated; verify before applying. React 👍/👎 to flag quality.

Copy link
Copy Markdown
Contributor

@hudi-agent hudi-agent May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 The current diff already calls super.close() at line 313, so the chaining concern looks addressed. The try/finally point still has merit though — if writeClient.close() throws, super.close() would be skipped. Worth noting that the sibling StreamWriteFunction.close() already uses this try/finally pattern for memorySegmentPool, so mirroring it here would be consistent.

this.writeClient.close();
}
super.close();
}
Copy link
Copy Markdown
Contributor

@hudi-agent hudi-agent May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 super.close() is invoked twice here (once before writeClient.close() and once after). Could you remove the duplicate trailing call? Depending on the parent's implementation, double-invoking could trigger errors or repeat side effects during teardown.

- AI-generated; verify before applying. React 👍/👎 to flag quality.

Copy link
Copy Markdown
Contributor

@hudi-agent hudi-agent May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Looking at the current diff, super.close() is only invoked once — at the end of the new close() method after writeClient.close(). There's no duplicate call. The try/finally suggestion from the sibling comment still seems worth considering though, so super.close() isn't skipped if writeClient.close() throws.


// -------------------------------------------------------------------------
// Getter/Setter
// -------------------------------------------------------------------------
Expand Down
Loading