danny0405 commented on code in PR #19399:
URL: https://github.com/apache/hudi/pull/19399#discussion_r3670917923
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/InsertFunctionWrapper.java:
##########
@@ -209,9 +211,22 @@ public AbstractWriteFunction getWriteFunction() {
@Override
public void close() throws Exception {
- this.coordinator.close();
- if (clusteringFunctionWrapper != null) {
- clusteringFunctionWrapper.close();
+ try {
+ if (writeFunction != null) {
+ writeFunction.close();
+ }
+ } finally {
+ try {
+ this.coordinator.close();
+ } finally {
+ try {
+ this.ioManager.close();
+ } finally {
+ if (clusteringFunctionWrapper != null) {
+ clusteringFunctionWrapper.close();
Review Comment:
[P1] This still does not close the async-clustering resources:
`ClusteringFunctionWrapper.close()` currently closes only its private
`IOManager`, not the `clusteringPlanOperator`, `clusteringOperator`, or
`commitSink` opened by `openFunction()`, so their write clients/executor remain
live. Also, `this.ioManager` above backs the `streamTask` passed to
`clusteringOperator`, yet it is closed before this call. Please make the
wrapper close the sink/operators (with null guards) and invoke it before
shutting down either IO manager.
--
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]
