hudi-agent commented on code in PR #20156:
URL: https://github.com/apache/hudi/pull/20156#discussion_r4141936373
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/CleanFunctionV2.java:
##########
@@ -111,13 +85,7 @@ public void initializeState(FunctionInitializationContext
context) throws Except
@Override
public void close() throws Exception {
- if (executor != null) {
- executor.close();
- }
-
- if (this.writeClient != null) {
- this.writeClient.close();
- }
+ this.cleanHandlerOpt.ifPresent(CleanHandler::close);
Review Comment:
🤖 If `open()` fails before `cleanHandlerOpt` is assigned (for example,
`createWriteClient` throws on a transient storage error), Flink still calls
`close()`, and this line now throws an NPE. The old V2 `close()` null-checked
`executor`/`writeClient`. Could this be guarded, e.g. `if (cleanHandlerOpt !=
null)`? `CompactionCommitSinkV2.close()` also reaches it through `finally {
super.close(); }`.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/MetadataTableCompactionCommitHandler.java:
##########
@@ -58,8 +58,23 @@
*/
public class MetadataTableCompactionCommitHandler extends
DataTableCompactionCommitHandler {
+ /** Data client retained for cleanup when no data-table handler owns it. */
+ private final HoodieFlinkWriteClient dataWriteClient;
Review Comment:
🤖 nit: the package-private 3-arg constructor takes a nullable
`dataWriteClient` whose null-ness encodes whether this handler owns cleanup of
the data client. Might be worth a clearer param name (e.g.
`dataClientToCloseOnShutdown`) or a one-line comment at each call site to make
that contract obvious.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/CleanFunctionV2.java:
##########
@@ -63,45 +59,23 @@ public CleanFunctionV2(Configuration conf) {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- this.writeClient = FlinkWriteClients.createWriteClient(conf,
getRuntimeContext());
- this.executor =
NonThrownExecutor.builder(log).waitForTasksFinish(true).build();
if (conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
- executor.execute(() -> {
- this.isCleaning = true;
- try {
- this.writeClient.clean();
- } finally {
- this.isCleaning = false;
- }
- }, "wait for cleaning finish");
+ HoodieFlinkWriteClient writeClient =
FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
+ this.cleanHandlerOpt =
Option.of(TableServiceHandlerFactory.createCleanHandler(conf, writeClient));
Review Comment:
🤖 Now that `writeClient` is a local, if `createCleanHandler` throws (say
`StreamerUtil.createMetadataWriteClient` fails with streaming index writes
enabled), nothing closes this client. The first `DefaultCleanHandler`'s
executor thread leaks too. Could `createCleanHandler` close what it created on
failure, the same way `createCompactionCommitHandler` now does? The same
applies to `CleanFunction`.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/TableServiceHandlerFactory.java:
##########
@@ -65,17 +66,25 @@ public static CompactHandler
createCompactHandler(Configuration conf, RuntimeCon
public static CompactionCommitHandler
createCompactionCommitHandler(Configuration conf, RuntimeContext
runtimeContext) {
HoodieFlinkWriteClient writeClient =
FlinkWriteClients.createWriteClient(conf, runtimeContext);
- boolean needsDataCompaction = OptionsResolver.needsAsyncCompaction(conf);
- boolean needsMetadataCompaction =
OptionsResolver.needsAsyncMetadataCompaction(conf);
- if (needsDataCompaction && needsMetadataCompaction) {
- return new CompositeCompactionCommitHandler(
- new DataTableCompactionCommitHandler(conf, writeClient),
- new MetadataTableCompactionCommitHandler(conf,
StreamerUtil.createMetadataWriteClient(writeClient)));
- }
- if (needsMetadataCompaction) {
- return new MetadataTableCompactionCommitHandler(conf,
StreamerUtil.createMetadataWriteClient(writeClient));
+ HoodieFlinkWriteClient metadataWriteClient = null;
+ try {
+ boolean needsDataCompaction = OptionsResolver.needsAsyncCompaction(conf);
+ boolean needsMetadataCompaction =
OptionsResolver.needsAsyncMetadataCompaction(conf);
+ if (needsMetadataCompaction) {
+ metadataWriteClient =
StreamerUtil.createMetadataWriteClient(writeClient);
+ if (needsDataCompaction) {
+ return new CompositeCompactionCommitHandler(
+ new DataTableCompactionCommitHandler(conf, writeClient),
+ new MetadataTableCompactionCommitHandler(conf,
metadataWriteClient));
+ }
+ return new MetadataTableCompactionCommitHandler(conf,
metadataWriteClient, writeClient);
Review Comment:
🤖 Good catch retaining the data client here. `createCompactionPlanHandler`
(line 47) and `createCompactHandler` (line 62) still have the same
metadata-only gap: the data `writeClient` is created, and only the MDT client
is passed to the handler. Is that worth fixing in this PR or a quick follow-up?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
