This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 669b4d861229 refactor(flink): share table-service handlers across sink
APIs (#20156)
669b4d861229 is described below
commit 669b4d86122979cd35840d5d88c7bc07b08825aa
Author: Danny Chan <[email protected]>
AuthorDate: Thu Oct 1 19:10:49 2026 +0800
refactor(flink): share table-service handlers across sink APIs (#20156)
* refactor(flink): share table-service handlers across sink APIs
* fix(flink): preserve compaction sink cleanup failures
---
.../java/org/apache/hudi/sink/CleanFunction.java | 13 +-
.../hudi/sink/clustering/ClusteringCommitSink.java | 14 +-
.../hudi/sink/compact/CompactionCommitSink.java | 23 +--
.../handler/DataTableCompactionCommitHandler.java | 15 ++
.../MetadataTableCompactionCommitHandler.java | 15 ++
.../handler/TableServiceHandlerFactory.java | 29 ++--
.../org/apache/hudi/sink/v2/CleanFunctionV2.java | 56 ++-----
.../sink/v2/clustering/ClusteringCommitSinkV2.java | 16 +-
.../sink/v2/compact/CompactionCommitSinkV2.java | 170 ++------------------
.../sink/clustering/TestClusteringCommitSink.java | 37 ++++-
.../sink/compact/CompactionCommitTestSink.java | 27 ++--
.../sink/compact/TestCompactionCommitSink.java | 87 +++++++++++
.../compact/handler/TestDefaultCleanHandler.java | 47 ++++++
.../handler/TestTableServiceHandlerFactory.java | 172 +++++++++++++++++++++
.../apache/hudi/sink/v2/TestCleanFunctionV2.java | 125 +++++++--------
.../v2/clustering/TestClusteringCommitSinkV2.java | 36 ++++-
.../v2/compact/TestCompactionCommitSinkV2.java | 106 ++++++++++++-
17 files changed, 653 insertions(+), 335 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
index 50be4ef5b79a..ca53365dc0fd 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
@@ -27,7 +27,6 @@ import org.apache.hudi.sink.compact.handler.CleanHandler;
import org.apache.hudi.sink.compact.handler.TableServiceHandlerFactory;
import org.apache.hudi.util.FlinkWriteClients;
-import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
@@ -41,14 +40,11 @@ import
org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
* at a time, a new task can not be scheduled until the last task
finished(fails or normally succeed).
* The cleaning task never expects to throw but only log.
*/
-@Slf4j
public class CleanFunction<T> extends AbstractRichFunctionAdapter
implements SinkFunctionAdapter<T>, CheckpointedFunction,
CheckpointListener {
private final Configuration conf;
- protected HoodieFlinkWriteClient writeClient;
-
private transient Option<CleanHandler> cleanHandlerOpt;
public CleanFunction(Configuration conf) {
@@ -58,9 +54,12 @@ public class CleanFunction<T> extends
AbstractRichFunctionAdapter
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- this.writeClient = FlinkWriteClients.createWriteClient(conf,
getRuntimeContext());
- this.cleanHandlerOpt = conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED)
- ? Option.of(TableServiceHandlerFactory.createCleanHandler(conf,
writeClient)) : Option.empty();
+ if (conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
+ HoodieFlinkWriteClient writeClient =
FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
+ this.cleanHandlerOpt =
Option.of(TableServiceHandlerFactory.createCleanHandler(conf, writeClient));
+ } else {
+ this.cleanHandlerOpt = Option.empty();
+ }
this.cleanHandlerOpt.ifPresent(CleanHandler::clean);
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
index 2bc668d2afa7..d9f16e8755d0 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
@@ -20,6 +20,7 @@ package org.apache.hudi.sink.clustering;
import org.apache.hudi.avro.model.HoodieClusteringGroup;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFileGroupId;
@@ -73,6 +74,8 @@ public class ClusteringCommitSink extends
CleanFunction<ClusteringCommitEvent> {
private transient HoodieFlinkTable<?> table;
+ protected transient HoodieFlinkWriteClient writeClient;
+
/**
* Buffer to collect the event from each clustering task {@code
ClusteringFunction}.
*
@@ -98,9 +101,7 @@ public class ClusteringCommitSink extends
CleanFunction<ClusteringCommitEvent> {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- if (writeClient == null) {
- this.writeClient = FlinkWriteClients.createWriteClient(conf,
getRuntimeContext());
- }
+ this.writeClient = FlinkWriteClients.createWriteClient(conf,
getRuntimeContext());
this.commitBuffer = new HashMap<>();
this.clusteringPlanCache = new HashMap<>();
this.table = writeClient.getHoodieTable();
@@ -122,6 +123,13 @@ public class ClusteringCommitSink extends
CleanFunction<ClusteringCommitEvent> {
commitIfNecessary(instant, commitBuffer.get(instant).values());
}
+ @Override
+ public void close() throws Exception {
+ try (HoodieFlinkWriteClient ignored = writeClient) {
+ super.close();
+ }
+ }
+
private long getNumErrorRecords(ClusteringCommitEvent event) {
if (event.getWriteStatuses() == null) {
return -1L;
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java
index 81b753d9ed3e..118c260f0b62 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java
@@ -19,13 +19,12 @@
package org.apache.hudi.sink.compact;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
-import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.sink.CleanFunction;
import org.apache.hudi.sink.compact.handler.CompactionCommitHandler;
import org.apache.hudi.sink.compact.handler.TableServiceHandlerFactory;
-import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.IOUtils;
/**
* Function to check and commit the compaction action.
@@ -38,7 +37,6 @@ import org.apache.flink.configuration.Configuration;
* <p>It also inherits the {@link CleanFunction} cleaning ability. This is
needed because
* the SQL API does not allow multiple sinks in one table sink provider.
*/
-@Slf4j
public class CompactionCommitSink extends CleanFunction<CompactionCommitEvent>
{
/**
@@ -62,28 +60,11 @@ public class CompactionCommitSink extends
CleanFunction<CompactionCommitEvent> {
@Override
public void invoke(CompactionCommitEvent event, Context context) throws
Exception {
- final String instant = event.getInstant();
- if (event.isFailed()
- || (event.getWriteStatuses() != null
- && event.getWriteStatuses().stream().anyMatch(writeStatus ->
writeStatus.getTotalErrorRecords() > 0))) {
- log.warn("Received abnormal CompactionCommitEvent of instant {}, task ID
is {},"
- + " is failed: {}, error record count: {}",
- instant, event.getTaskID(), event.isFailed(),
getNumErrorRecords(event));
- }
compactCommitHandler.commitIfNecessary(event);
}
@Override
public void close() throws Exception {
- compactCommitHandler.close();
- super.close();
- }
-
- private long getNumErrorRecords(CompactionCommitEvent event) {
- if (event.getWriteStatuses() == null) {
- return -1L;
- }
- return event.getWriteStatuses().stream()
- .map(WriteStatus::getTotalErrorRecords).reduce(Long::sum).orElse(0L);
+ IOUtils.closeAll(compactCommitHandler, super::close);
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/DataTableCompactionCommitHandler.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/DataTableCompactionCommitHandler.java
index 2b3d7de9437d..717a0b4b29de 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/DataTableCompactionCommitHandler.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/DataTableCompactionCommitHandler.java
@@ -111,6 +111,13 @@ public class DataTableCompactionCommitHandler implements
CompactionCommitHandler
@Override
public void commitIfNecessary(CompactionCommitEvent event) {
String instant = event.getInstant();
+ if (event.isFailed()
+ || (event.getWriteStatuses() != null
+ && event.getWriteStatuses().stream().anyMatch(writeStatus ->
writeStatus.getTotalErrorRecords() > 0))) {
+ log.warn("Received abnormal CompactionCommitEvent of instant {}, task ID
is {},"
+ + " is failed: {}, error record count: {}",
+ instant, event.getTaskID(), event.isFailed(),
getNumErrorRecords(event));
+ }
commitBuffer.computeIfAbsent(instant, k -> new HashMap<>())
.put(event.getFileId(), event);
@@ -253,6 +260,14 @@ public class DataTableCompactionCommitHandler implements
CompactionCommitHandler
this.compactionPlanCache.remove(instant);
}
+ private long getNumErrorRecords(CompactionCommitEvent event) {
+ if (event.getWriteStatuses() == null) {
+ return -1L;
+ }
+ return event.getWriteStatuses().stream()
+ .map(WriteStatus::getTotalErrorRecords).reduce(Long::sum).orElse(0L);
+ }
+
@Override
public void close() {
writeClient.close();
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/MetadataTableCompactionCommitHandler.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/MetadataTableCompactionCommitHandler.java
index ced1b0a3a6da..4ff0a9d6228b 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/MetadataTableCompactionCommitHandler.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/MetadataTableCompactionCommitHandler.java
@@ -58,8 +58,23 @@ import java.util.List;
*/
public class MetadataTableCompactionCommitHandler extends
DataTableCompactionCommitHandler {
+ /** Data client retained for cleanup when no data-table handler owns it. */
+ private final HoodieFlinkWriteClient dataWriteClient;
+
public MetadataTableCompactionCommitHandler(Configuration conf,
HoodieFlinkWriteClient writeClient) {
+ this(conf, writeClient, null);
+ }
+
+ MetadataTableCompactionCommitHandler(Configuration conf,
HoodieFlinkWriteClient writeClient, HoodieFlinkWriteClient dataWriteClient) {
super(conf, writeClient);
+ this.dataWriteClient = dataWriteClient;
+ }
+
+ @Override
+ public void close() {
+ try (HoodieFlinkWriteClient ignored = dataWriteClient) {
+ super.close();
+ }
}
@Override
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/TableServiceHandlerFactory.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/TableServiceHandlerFactory.java
index 2965866094d1..a7515b966784 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/TableServiceHandlerFactory.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/TableServiceHandlerFactory.java
@@ -19,6 +19,7 @@
package org.apache.hudi.sink.compact.handler;
import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.common.util.CloseableUtils;
import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.util.FlinkWriteClients;
import org.apache.hudi.util.StreamerUtil;
@@ -65,17 +66,25 @@ public final class TableServiceHandlerFactory {
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);
+ }
+ return new DataTableCompactionCommitHandler(conf, writeClient);
+ } catch (RuntimeException | Error failure) {
+ CloseableUtils.closeSuppressing(metadataWriteClient, failure);
+ CloseableUtils.closeSuppressing(writeClient, failure);
+ throw failure;
}
- return new DataTableCompactionCommitHandler(conf, writeClient);
}
public static CleanHandler createCleanHandler(Configuration conf,
HoodieFlinkWriteClient writeClient) {
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/CleanFunctionV2.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/CleanFunctionV2.java
index 93d7496debb0..69c4f217ef64 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/CleanFunctionV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/CleanFunctionV2.java
@@ -20,11 +20,12 @@ package org.apache.hudi.sink.v2;
import org.apache.hudi.adapter.ProcessFunctionAdapter;
import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.configuration.FlinkOptions;
-import org.apache.hudi.sink.utils.NonThrownExecutor;
+import org.apache.hudi.sink.compact.handler.CleanHandler;
+import org.apache.hudi.sink.compact.handler.TableServiceHandlerFactory;
import org.apache.hudi.util.FlinkWriteClients;
-import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
@@ -44,17 +45,12 @@ import org.apache.flink.util.Collector;
* <p>Note: The difference with {@code CleanFunction} is {@code
CleanFunctionV2} extends {@code ProcessFunction},
* while {@code CleanFunction} is a {@code SinkFunction}.
*/
-@Slf4j
public class CleanFunctionV2<T> extends ProcessFunctionAdapter<T, RowData>
implements CheckpointedFunction, CheckpointListener {
private final Configuration conf;
- protected HoodieFlinkWriteClient writeClient;
-
- private NonThrownExecutor executor;
-
- protected volatile boolean isCleaning;
+ private transient Option<CleanHandler> cleanHandlerOpt;
public CleanFunctionV2(Configuration conf) {
this.conf = conf;
@@ -63,45 +59,23 @@ public class CleanFunctionV2<T> extends
ProcessFunctionAdapter<T, RowData>
@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));
+ } else {
+ this.cleanHandlerOpt = Option.empty();
}
+ this.cleanHandlerOpt.ifPresent(CleanHandler::clean);
}
@Override
public void notifyCheckpointComplete(long l) throws Exception {
- if (conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED) && isCleaning) {
- executor.execute(() -> {
- try {
- this.writeClient.waitForCleaningFinish();
- } finally {
- // ensure to switch the isCleaning flag
- this.isCleaning = false;
- }
- }, "wait for cleaning finish");
- }
+ this.cleanHandlerOpt.ifPresent(CleanHandler::waitForCleaningFinish);
}
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
- if (conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
- try {
- this.writeClient.startAsyncCleaning();
- this.isCleaning = true;
- } catch (Throwable throwable) {
- // catch the exception to not affect the normal checkpointing
- log.warn("Unable to start async cleaning", throwable);
- }
- }
+ this.cleanHandlerOpt.ifPresent(CleanHandler::startAsyncCleaning);
}
@Override
@@ -111,13 +85,7 @@ public class CleanFunctionV2<T> extends
ProcessFunctionAdapter<T, RowData>
@Override
public void close() throws Exception {
- if (executor != null) {
- executor.close();
- }
-
- if (this.writeClient != null) {
- this.writeClient.close();
- }
+ this.cleanHandlerOpt.ifPresent(CleanHandler::close);
}
@Override
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java
index 4f31ee877f7a..51148e6085d8 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java
@@ -20,6 +20,7 @@ package org.apache.hudi.sink.v2.clustering;
import org.apache.hudi.avro.model.HoodieClusteringGroup;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
+import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFileGroupId;
@@ -80,6 +81,8 @@ public class ClusteringCommitSinkV2 extends
CleanFunctionV2<ClusteringCommitEven
private transient HoodieFlinkTable<?> table;
+ protected transient HoodieFlinkWriteClient writeClient;
+
/**
* Buffer to collect the event from each clustering task {@code
ClusteringFunction}.
*
@@ -105,9 +108,7 @@ public class ClusteringCommitSinkV2 extends
CleanFunctionV2<ClusteringCommitEven
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- if (writeClient == null) {
- this.writeClient = FlinkWriteClients.createWriteClient(conf,
getRuntimeContext());
- }
+ this.writeClient = FlinkWriteClients.createWriteClient(conf,
getRuntimeContext());
this.commitBuffer = new HashMap<>();
this.clusteringPlanCache = new HashMap<>();
this.table = writeClient.getHoodieTable();
@@ -132,6 +133,13 @@ public class ClusteringCommitSinkV2 extends
CleanFunctionV2<ClusteringCommitEven
commitIfNecessary(instant, commitBuffer.get(instant).values());
}
+ @Override
+ public void close() throws Exception {
+ try (HoodieFlinkWriteClient ignored = writeClient) {
+ super.close();
+ }
+ }
+
private long getNumErrorRecords(ClusteringCommitEvent event) {
if (event.getWriteStatuses() == null) {
return -1L;
@@ -230,7 +238,7 @@ public class ClusteringCommitSinkV2 extends
CleanFunctionV2<ClusteringCommitEven
clusteringMetrics.updateCommitMetrics(instant,
writeMetadata.getCommitMetadata().get());
// whether to clean up the input base parquet files used for clustering
- if (!conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
+ if (!conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
log.info("Running inline clean");
this.writeClient.clean();
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java
index 7e81112bb0b6..a8157e0e7da8 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java
@@ -19,33 +19,16 @@
package org.apache.hudi.sink.v2.compact;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
-import org.apache.hudi.client.WriteStatus;
-import org.apache.hudi.common.data.HoodieListData;
-import org.apache.hudi.common.model.HoodieCommitMetadata;
-import org.apache.hudi.common.util.CompactionUtils;
-import org.apache.hudi.configuration.FlinkOptions;
-import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.metrics.FlinkCompactionMetrics;
import org.apache.hudi.sink.compact.CompactionCommitEvent;
+import org.apache.hudi.sink.compact.handler.CompactionCommitHandler;
+import org.apache.hudi.sink.compact.handler.TableServiceHandlerFactory;
import org.apache.hudi.sink.v2.CleanFunctionV2;
-import org.apache.hudi.table.HoodieFlinkTable;
-import org.apache.hudi.table.action.compact.CompactHelpers;
-import org.apache.hudi.util.CompactionUtil;
-import org.apache.hudi.util.FlinkWriteClients;
-import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.data.RowData;
import org.apache.flink.util.Collector;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
+import org.apache.flink.util.IOUtils;
/**
* Function to check and commit the compaction action.
@@ -61,7 +44,6 @@ import java.util.stream.Collectors;
* <p>Note: The difference with {@code CompactionCommitSink} is {@code
CompactionCommitSinkV2}
* extends {@code ProcessFunction}, while {@code CompactionCommitSink} is a
{@code SinkFunction}.
*/
-@Slf4j
public class CompactionCommitSinkV2 extends
CleanFunctionV2<CompactionCommitEvent> {
/**
@@ -69,30 +51,7 @@ public class CompactionCommitSinkV2 extends
CleanFunctionV2<CompactionCommitEven
*/
private final Configuration conf;
- /**
- * Buffer to collect the event from each compact task {@code
CompactFunction}.
- *
- * <p>Stores the mapping of instant_time -> file_id -> event. Use a map to
collect the
- * events because the rolling back of intermediate compaction tasks
generates corrupt
- * events.
- */
- private transient Map<String, Map<String, CompactionCommitEvent>>
commitBuffer;
-
- /**
- * Cache to store compaction plan for each instant.
- * Stores the mapping of instant_time -> compactionPlan.
- */
- private transient Map<String, HoodieCompactionPlan> compactionPlanCache;
-
- /**
- * The hoodie table.
- */
- private transient HoodieFlinkTable<?> table;
-
- /**
- * Compaction metrics.
- */
- private transient FlinkCompactionMetrics compactionMetrics;
+ private transient CompactionCommitHandler compactCommitHandler;
public CompactionCommitSinkV2(Configuration conf) {
super(conf);
@@ -102,13 +61,8 @@ public class CompactionCommitSinkV2 extends
CleanFunctionV2<CompactionCommitEven
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- if (writeClient == null) {
- this.writeClient = FlinkWriteClients.createWriteClient(conf,
getRuntimeContext());
- }
- this.commitBuffer = new HashMap<>();
- this.compactionPlanCache = new HashMap<>();
- this.table = this.writeClient.getHoodieTable();
- registerMetrics();
+ this.compactCommitHandler =
TableServiceHandlerFactory.createCompactionCommitHandler(conf,
getRuntimeContext());
+
this.compactCommitHandler.registerMetrics(getRuntimeContext().getMetricGroup());
}
@Override
@@ -116,115 +70,11 @@ public class CompactionCommitSinkV2 extends
CleanFunctionV2<CompactionCommitEven
CompactionCommitEvent event,
ProcessFunction<CompactionCommitEvent, RowData>.Context context,
Collector<RowData> collector) throws Exception {
- final String instant = event.getInstant();
- if (event.isFailed()
- || (event.getWriteStatuses() != null
- && event.getWriteStatuses().stream().anyMatch(writeStatus ->
writeStatus.getTotalErrorRecords() > 0))) {
- log.warn("Received abnormal CompactionCommitEvent of instant {}, task ID
is {},"
- + " is failed: {}, error record count: {}",
- instant, event.getTaskID(), event.isFailed(),
getNumErrorRecords(event));
- }
- commitBuffer.computeIfAbsent(instant, k -> new HashMap<>())
- .put(event.getFileId(), event);
- commitIfNecessary(instant, commitBuffer.get(instant).values());
- }
-
- private long getNumErrorRecords(CompactionCommitEvent event) {
- if (event.getWriteStatuses() == null) {
- return -1L;
- }
- return event.getWriteStatuses().stream()
- .map(WriteStatus::getTotalErrorRecords).reduce(Long::sum).orElse(0L);
- }
-
- /**
- * Condition to commit: the commit buffer has equal size with the compaction
plan operations
- * and all the compact commit event {@link CompactionCommitEvent} has the
same compaction instant time.
- *
- * @param instant Compaction commit instant time
- * @param events Commit events ever received for the instant
- */
- private void commitIfNecessary(String instant,
Collection<CompactionCommitEvent> events) throws IOException {
- HoodieCompactionPlan compactionPlan =
compactionPlanCache.computeIfAbsent(instant, k -> {
- try {
- return CompactionUtils.getCompactionPlan(
- this.writeClient.getHoodieTable().getMetaClient(), instant);
- } catch (Exception e) {
- throw new HoodieException(e);
- }
- });
-
- boolean isReady = compactionPlan.getOperations().size() == events.size();
- if (!isReady) {
- return;
- }
-
- if (events.stream().anyMatch(CompactionCommitEvent::isFailed)) {
- try {
- // handle failure case
- CompactionUtil.rollbackCompaction(table, instant,
writeClient.getTransactionManager());
- } finally {
- // remove commitBuffer to avoid obsolete metadata commit
- reset(instant);
- this.compactionMetrics.markCompactionRolledBack();
- }
- return;
- }
-
- try {
- doCommit(instant, events);
- } catch (Throwable throwable) {
- // make it fail-safe
- log.error("Error while committing compaction instant: {}", instant,
throwable);
- this.compactionMetrics.markCompactionRolledBack();
- } finally {
- // reset the status
- reset(instant);
- }
- }
-
- @SuppressWarnings("unchecked")
- private void doCommit(String instant, Collection<CompactionCommitEvent>
events) throws IOException {
- List<WriteStatus> statuses = events.stream()
- .map(CompactionCommitEvent::getWriteStatuses)
- .flatMap(Collection::stream)
- .collect(Collectors.toList());
-
- long numErrorRecords =
statuses.stream().map(WriteStatus::getTotalErrorRecords).reduce(Long::sum).orElse(0L);
-
- if (numErrorRecords > 0 && !this.conf.get(FlinkOptions.IGNORE_FAILED)) {
- // handle failure case
- log.error("Got {} error records during compaction of instant {},\n"
- + "option '{}' is configured as false,"
- + "rolls back the compaction", numErrorRecords, instant,
FlinkOptions.IGNORE_FAILED.key());
- CompactionUtil.rollbackCompaction(table, instant,
writeClient.getTransactionManager());
- this.compactionMetrics.markCompactionRolledBack();
- return;
- }
-
- HoodieCommitMetadata metadata =
CompactHelpers.getInstance().createCompactionMetadata(
- table, instant, HoodieListData.eager(statuses),
writeClient.getConfig().getSchema());
-
- // commit the compaction
- this.writeClient.completeCompaction(metadata, table, instant);
-
- this.compactionMetrics.updateCommitMetrics(instant, metadata);
- this.compactionMetrics.markCompactionCompleted();
-
- // Whether to clean up the old log file when compaction
- if (!conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
- this.writeClient.clean();
- }
- }
-
- private void reset(String instant) {
- this.commitBuffer.remove(instant);
- this.compactionPlanCache.remove(instant);
+ compactCommitHandler.commitIfNecessary(event);
}
- private void registerMetrics() {
- MetricGroup metrics = getRuntimeContext().getMetricGroup();
- compactionMetrics = new FlinkCompactionMetrics(metrics);
- compactionMetrics.registerMetrics();
+ @Override
+ public void close() throws Exception {
+ IOUtils.closeAll(compactCommitHandler, super::close);
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/clustering/TestClusteringCommitSink.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/clustering/TestClusteringCommitSink.java
index e3486f3d3cc5..9e55ee0d9999 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/clustering/TestClusteringCommitSink.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/clustering/TestClusteringCommitSink.java
@@ -43,15 +43,19 @@ import org.apache.hudi.util.FlinkWriteClients;
import org.apache.flink.configuration.Configuration;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.MockedStatic;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
@@ -68,6 +72,7 @@ class TestClusteringCommitSink {
private Configuration conf;
private HoodieFlinkWriteClient writeClient;
+ private HoodieFlinkWriteClient cleaningClient;
private HoodieFlinkTable table;
private HoodieTableMetaClient metaClient;
private HoodieActiveTimeline activeTimeline;
@@ -80,6 +85,7 @@ class TestClusteringCommitSink {
conf = new Configuration();
conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, false);
writeClient = mock(HoodieFlinkWriteClient.class);
+ cleaningClient = mock(HoodieFlinkWriteClient.class);
table = mock(HoodieFlinkTable.class);
metaClient = mock(HoodieTableMetaClient.class);
activeTimeline = mock(HoodieActiveTimeline.class);
@@ -183,8 +189,10 @@ class TestClusteringCommitSink {
}
}
- @Test
- void testCommitSuccessfulClusteringAndCleanInline() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testCommitSuccessfulClusteringAndCloseClients(boolean asyncCleaning)
throws Exception {
+ conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, asyncCleaning);
ClusteringCommitSink sink = openSink();
HoodieClusteringPlan plan = planWithGroups(1);
HoodieWriteStat writeStat = new HoodieWriteStat();
@@ -213,7 +221,21 @@ class TestClusteringCommitSink {
any(HoodieCommitMetadata.class),
same(table),
eq(INSTANT));
- verify(writeClient).clean();
+ sink.close();
+ verify(writeClient, times(asyncCleaning ? 0 : 1)).clean();
+ verify(writeClient).close();
+ verify(cleaningClient, times(asyncCleaning ? 1 : 0)).clean();
+ verify(cleaningClient, times(asyncCleaning ? 1 : 0)).close();
+ }
+
+ @Test
+ void testClosesClusteringClientWhenCleaningClientCloseFails() throws
Exception {
+ conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, true);
+ doThrow(new RuntimeException("expected")).when(cleaningClient).close();
+ ClusteringCommitSink sink = openSink();
+ assertThrows(RuntimeException.class, sink::close);
+ verify(cleaningClient).close();
+ verify(writeClient).close();
}
private ClusteringCommitSink openSink() throws Exception {
@@ -221,9 +243,14 @@ class TestClusteringCommitSink {
MockStreamingRuntimeContext runtimeContext = new
MockStreamingRuntimeContext(false, 1, 0);
sink.setRuntimeContext(runtimeContext);
try (MockedStatic<FlinkWriteClients> writeClients =
mockStatic(FlinkWriteClients.class)) {
- writeClients.when(() -> FlinkWriteClients.createWriteClient(conf,
runtimeContext))
- .thenReturn(writeClient);
+ if (conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(conf,
runtimeContext)).thenReturn(cleaningClient, writeClient);
+ } else {
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(conf,
runtimeContext)).thenReturn(writeClient);
+ }
sink.open(conf);
+ writeClients.verify(() -> FlinkWriteClients.createWriteClient(conf,
runtimeContext),
+ times(conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED) ? 2 : 1));
}
return sink;
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/CompactionCommitTestSink.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/CompactionCommitTestSink.java
index e9f8f72f593f..cb2057c1506d 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/CompactionCommitTestSink.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/CompactionCommitTestSink.java
@@ -18,12 +18,14 @@
package org.apache.hudi.sink.compact;
+import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.client.common.HoodieFlinkEngineContext;
import org.apache.hudi.client.timeline.HoodieTimelineArchiver;
import org.apache.hudi.client.timeline.TimelineArchiverV2;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.config.HoodieArchivalConfig;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.util.FlinkWriteClients;
import org.apache.hudi.utils.RuntimeContextUtils;
import org.apache.flink.configuration.Configuration;
@@ -34,23 +36,28 @@ import java.util.List;
* CompactionCommitTestSink, throw for first attempt to simulate failure
*/
public class CompactionCommitTestSink extends CompactionCommitSink {
+ private final Configuration conf;
+
public CompactionCommitTestSink(Configuration conf) {
super(conf);
+ this.conf = conf;
}
@Override
public void invoke(CompactionCommitEvent event, Context context) throws
Exception {
super.invoke(event, context);
- List<HoodieInstant> instants =
writeClient.getHoodieTable().getMetaClient().getActiveTimeline().getInstants();
- boolean compactCommitted = instants.stream().anyMatch(i ->
i.requestedTime().equals(event.getInstant()) && i.isCompleted());
- if (compactCommitted &&
RuntimeContextUtils.getAttemptNumber(getRuntimeContext()) == 0) {
- // archive compact instant
-
this.writeClient.getConfig().setValue(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP,
"1");
-
this.writeClient.getConfig().setValue(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP,
"1");
- HoodieTimelineArchiver archiver = new
TimelineArchiverV2(this.writeClient.getConfig(),
this.writeClient.getHoodieTable());
- this.writeClient.getHoodieTable().getMetaClient().reloadActiveTimeline();
- archiver.archiveIfRequired(HoodieFlinkEngineContext.DEFAULT);
- throw new HoodieException("Fail first attempt to simulate failover in
test.");
+ try (HoodieFlinkWriteClient writeClient =
FlinkWriteClients.createWriteClient(conf, getRuntimeContext())) {
+ List<HoodieInstant> instants =
writeClient.getHoodieTable().getMetaClient().getActiveTimeline().getInstants();
+ boolean compactCommitted = instants.stream().anyMatch(i ->
i.requestedTime().equals(event.getInstant()) && i.isCompleted());
+ if (compactCommitted &&
RuntimeContextUtils.getAttemptNumber(getRuntimeContext()) == 0) {
+ // archive compact instant
+
writeClient.getConfig().setValue(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP, "1");
+
writeClient.getConfig().setValue(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP, "1");
+ HoodieTimelineArchiver archiver = new
TimelineArchiverV2(writeClient.getConfig(), writeClient.getHoodieTable());
+ writeClient.getHoodieTable().getMetaClient().reloadActiveTimeline();
+ archiver.archiveIfRequired(HoodieFlinkEngineContext.DEFAULT);
+ throw new HoodieException("Fail first attempt to simulate failover in
test.");
+ }
}
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/TestCompactionCommitSink.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/TestCompactionCommitSink.java
new file mode 100644
index 000000000000..2a2ecff94fef
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/TestCompactionCommitSink.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.compact;
+
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.sink.compact.handler.CleanHandler;
+import org.apache.hudi.sink.compact.handler.CompactionCommitHandler;
+import org.apache.hudi.sink.compact.handler.TableServiceHandlerFactory;
+import org.apache.hudi.sink.utils.MockStreamingRuntimeContext;
+import org.apache.hudi.sink.v2.compact.CompactionCommitSinkV2;
+import org.apache.hudi.util.FlinkWriteClients;
+
+import org.apache.flink.configuration.Configuration;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.InOrder;
+import org.mockito.MockedStatic;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+
+/** Tests cleanup failure handling across both compaction sink APIs. */
+class TestCompactionCommitSink {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testClosePreservesCommitFailureAndSuppressesCleaningFailure(boolean
useSinkV2) throws Exception {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, true);
+ MockStreamingRuntimeContext runtimeContext = new
MockStreamingRuntimeContext(false, 1, 0);
+ HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
+ CompactionCommitHandler commitHandler =
mock(CompactionCommitHandler.class);
+ CleanHandler cleanHandler = mock(CleanHandler.class);
+ RuntimeException commitFailure = new RuntimeException("commit handler
close failed");
+ RuntimeException cleanFailure = new RuntimeException("clean handler close
failed");
+ doThrow(commitFailure).when(commitHandler).close();
+ doThrow(cleanFailure).when(cleanHandler).close();
+
+ try (MockedStatic<FlinkWriteClients> clients =
mockStatic(FlinkWriteClients.class);
+ MockedStatic<TableServiceHandlerFactory> factory =
mockStatic(TableServiceHandlerFactory.class)) {
+ clients.when(() -> FlinkWriteClients.createWriteClient(conf,
runtimeContext)).thenReturn(writeClient);
+ factory.when(() -> TableServiceHandlerFactory.createCleanHandler(conf,
writeClient)).thenReturn(cleanHandler);
+ factory.when(() ->
TableServiceHandlerFactory.createCompactionCommitHandler(conf,
runtimeContext)).thenReturn(commitHandler);
+
+ AutoCloseable sink;
+ if (useSinkV2) {
+ CompactionCommitSinkV2 function = new CompactionCommitSinkV2(conf);
+ function.setRuntimeContext(runtimeContext);
+ function.open(conf);
+ sink = function::close;
+ } else {
+ CompactionCommitSink function = new CompactionCommitSink(conf);
+ function.setRuntimeContext(runtimeContext);
+ function.open(conf);
+ sink = function::close;
+ }
+
+ assertSame(commitFailure, assertThrows(RuntimeException.class,
sink::close));
+ assertArrayEquals(new Throwable[] {cleanFailure},
commitFailure.getSuppressed());
+ InOrder order = inOrder(commitHandler, cleanHandler);
+ order.verify(commitHandler).close();
+ order.verify(cleanHandler).close();
+ }
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestDefaultCleanHandler.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestDefaultCleanHandler.java
index 7c2c5524d49d..56ff5260f3bb 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestDefaultCleanHandler.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestDefaultCleanHandler.java
@@ -21,13 +21,60 @@ package org.apache.hudi.sink.compact.handler;
import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
class TestDefaultCleanHandler {
+ @Test
+ void testCloseWaitsForInitialCleaning() {
+ HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
+ try (DefaultCleanHandler handler = new DefaultCleanHandler(writeClient)) {
+ handler.clean();
+ }
+ InOrder order = inOrder(writeClient);
+ order.verify(writeClient).clean();
+ order.verify(writeClient).close();
+ verifyNoMoreInteractions(writeClient);
+ }
+
+ @Test
+ void testCheckpointCleaningDoesNotStartTwice() {
+ HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
+ try (DefaultCleanHandler handler = new DefaultCleanHandler(writeClient)) {
+ handler.startAsyncCleaning();
+ handler.startAsyncCleaning();
+ handler.waitForCleaningFinish();
+ }
+ InOrder order = inOrder(writeClient);
+ order.verify(writeClient).startAsyncCleaning();
+ order.verify(writeClient).waitForCleaningFinish();
+ order.verify(writeClient).close();
+ verifyNoMoreInteractions(writeClient);
+ }
+
+ @Test
+ void testCleaningStartFailureAllowsRetry() {
+ HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
+ doThrow(new
RuntimeException("expected")).doNothing().when(writeClient).startAsyncCleaning();
+ try (DefaultCleanHandler handler = new DefaultCleanHandler(writeClient)) {
+ assertDoesNotThrow(handler::startAsyncCleaning);
+ handler.startAsyncCleaning();
+ handler.waitForCleaningFinish();
+ }
+ verify(writeClient, times(2)).startAsyncCleaning();
+ verify(writeClient).waitForCleaningFinish();
+ verify(writeClient).close();
+ verifyNoMoreInteractions(writeClient);
+ }
+
@Test
void testCloseClosesWriteClientWithoutTriggeringClean() {
HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestTableServiceHandlerFactory.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestTableServiceHandlerFactory.java
new file mode 100644
index 000000000000..4d7a84236c81
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestTableServiceHandlerFactory.java
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.compact.handler;
+
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.table.HoodieFlinkTable;
+import org.apache.hudi.util.FlinkWriteClients;
+import org.apache.hudi.util.StreamerUtil;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedStatic;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/** Tests write-client ownership in the compaction commit handler factory. */
+class TestTableServiceHandlerFactory {
+ private Configuration conf;
+ private RuntimeContext runtimeContext;
+ private HoodieFlinkWriteClient dataClient;
+ private HoodieFlinkWriteClient metadataClient;
+
+ @BeforeEach
+ @SuppressWarnings("unchecked")
+ void setUp() {
+ conf = new Configuration();
+ runtimeContext = mock(RuntimeContext.class);
+ dataClient = mock(HoodieFlinkWriteClient.class);
+ metadataClient = mock(HoodieFlinkWriteClient.class);
+ when(dataClient.getHoodieTable()).thenReturn(mock(HoodieFlinkTable.class));
+
when(metadataClient.getHoodieTable()).thenReturn(mock(HoodieFlinkTable.class));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false,false", "true,false", "false,true", "true,true"})
+ void testReturnedHandlerOwnsCreatedClients(boolean dataCompaction, boolean
metadataCompaction) {
+ configureCompaction(dataCompaction, metadataCompaction);
+ try (CompactionCommitHandler handler = createHandler()) {
+ if (dataCompaction && metadataCompaction) {
+ assertInstanceOf(CompositeCompactionCommitHandler.class, handler);
+ } else if (metadataCompaction) {
+ assertInstanceOf(MetadataTableCompactionCommitHandler.class, handler);
+ } else {
+ assertInstanceOf(DataTableCompactionCommitHandler.class, handler);
+ }
+ verify(dataClient, never()).close();
+ verify(metadataClient, never()).close();
+ }
+ verify(dataClient).close();
+ if (metadataCompaction) {
+ verify(metadataClient).close();
+ } else {
+ verifyNoInteractions(metadataClient);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMetadataClientCreationFailureClosesDataClient(boolean
dataCompaction) {
+ configureCompaction(dataCompaction, true);
+ RuntimeException failure = new RuntimeException("metadata client creation
failed");
+ RuntimeException closeFailure = new RuntimeException("data client close
failed");
+ doThrow(closeFailure).when(dataClient).close();
+ try (MockedStatic<FlinkWriteClients> writeClients =
mockStatic(FlinkWriteClients.class);
+ MockedStatic<StreamerUtil> streamerUtil =
mockStatic(StreamerUtil.class)) {
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(conf,
runtimeContext)).thenReturn(dataClient);
+ streamerUtil.when(() ->
StreamerUtil.createMetadataWriteClient(dataClient)).thenThrow(failure);
+
+ assertSame(failure, assertThrows(RuntimeException.class,
+ () -> TableServiceHandlerFactory.createCompactionCommitHandler(conf,
runtimeContext)));
+ }
+ verify(dataClient).close();
+ verifyNoInteractions(metadataClient);
+ assertArrayEquals(new Throwable[] {closeFailure}, failure.getSuppressed());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMetadataHandlerConstructionFailureClosesBothClients(boolean
dataCompaction) {
+ configureCompaction(dataCompaction, true);
+ RuntimeException failure = new RuntimeException("metadata handler
construction failed");
+ RuntimeException metadataCloseFailure = new RuntimeException("metadata
client close failed");
+ RuntimeException dataCloseFailure = new RuntimeException("data client
close failed");
+ when(metadataClient.getHoodieTable()).thenThrow(failure);
+ doThrow(metadataCloseFailure).when(metadataClient).close();
+ doThrow(dataCloseFailure).when(dataClient).close();
+
+ assertSame(failure, assertThrows(RuntimeException.class,
this::createHandler));
+
+ verify(metadataClient).close();
+ verify(dataClient).close();
+ assertArrayEquals(new Throwable[] {metadataCloseFailure,
dataCloseFailure}, failure.getSuppressed());
+ }
+
+ @Test
+ void testDataHandlerConstructionFailureClosesDataClient() {
+ configureCompaction(true, false);
+ RuntimeException failure = new RuntimeException("data handler construction
failed");
+ when(dataClient.getHoodieTable()).thenThrow(failure);
+
+ assertSame(failure, assertThrows(RuntimeException.class,
this::createHandler));
+
+ verify(dataClient).close();
+ verifyNoInteractions(metadataClient);
+ }
+
+ @Test
+ void testMetadataOnlyCloseFailureStillClosesDataClient() {
+ configureCompaction(false, true);
+ CompactionCommitHandler handler = createHandler();
+ RuntimeException failure = new RuntimeException("metadata client close
failed");
+ RuntimeException closeFailure = new RuntimeException("data client close
failed");
+ doThrow(failure).when(metadataClient).close();
+ doThrow(closeFailure).when(dataClient).close();
+
+ assertSame(failure, assertThrows(RuntimeException.class, handler::close));
+
+ verify(metadataClient).close();
+ verify(dataClient).close();
+ assertArrayEquals(new Throwable[] {closeFailure}, failure.getSuppressed());
+ }
+
+ private CompactionCommitHandler createHandler() {
+ try (MockedStatic<FlinkWriteClients> writeClients =
mockStatic(FlinkWriteClients.class);
+ MockedStatic<StreamerUtil> streamerUtil =
mockStatic(StreamerUtil.class)) {
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(conf,
runtimeContext)).thenReturn(dataClient);
+ streamerUtil.when(() ->
StreamerUtil.createMetadataWriteClient(dataClient)).thenReturn(metadataClient);
+ return TableServiceHandlerFactory.createCompactionCommitHandler(conf,
runtimeContext);
+ }
+ }
+
+ private void configureCompaction(boolean dataCompaction, boolean
metadataCompaction) {
+ conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
+ conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, dataCompaction);
+ conf.set(FlinkOptions.METADATA_ENABLED, metadataCompaction);
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.GLOBAL_RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.METADATA_COMPACTION_ASYNC_ENABLED,
metadataCompaction);
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/TestCleanFunctionV2.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/TestCleanFunctionV2.java
index ca2d75cd3ff0..164aede00e9b 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/TestCleanFunctionV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/TestCleanFunctionV2.java
@@ -20,7 +20,12 @@ package org.apache.hudi.sink.v2;
import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.sink.compact.handler.CleanHandler;
+import org.apache.hudi.sink.compact.handler.DefaultCleanHandler;
+import org.apache.hudi.sink.compact.handler.TableServiceHandlerFactory;
import org.apache.hudi.util.FlinkWriteClients;
+import org.apache.hudi.util.StreamerUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.operators.ProcessOperator;
@@ -28,21 +33,23 @@ import
org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.RowData;
import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
-import java.util.concurrent.Callable;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
/** Tests checkpoint-driven cleaning in {@link CleanFunctionV2}. */
class TestCleanFunctionV2 {
@@ -52,46 +59,54 @@ class TestCleanFunctionV2 {
Configuration conf = new Configuration();
conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, true);
HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
- CleanFunctionV2<String> function = new CleanFunctionV2<>(conf);
-
- try (OneInputStreamOperatorTestHarness<String, RowData> harness =
- openHarness(conf, function, writeClient)) {
- // Opening waits for any cleaning left by a previous job attempt.
- verify(writeClient, timeout(5000)).clean();
- assertTrue(waitUntil(() -> !function.isCleaning));
-
- harness.processElement(new StreamRecord<>("ignored"));
- assertTrue(harness.getOutput().isEmpty());
-
- harness.snapshot(1, 1);
- verify(writeClient).startAsyncCleaning();
- assertTrue(function.isCleaning);
-
- harness.notifyOfCompletedCheckpoint(1);
- verify(writeClient, timeout(5000)).waitForCleaningFinish();
- assertTrue(waitUntil(() -> !function.isCleaning));
+ CleanHandler cleanHandler = mock(CleanHandler.class);
+
+ try (MockedStatic<TableServiceHandlerFactory> factory =
mockStatic(TableServiceHandlerFactory.class)) {
+ factory.when(() -> TableServiceHandlerFactory.createCleanHandler(conf,
writeClient)).thenReturn(cleanHandler);
+ try (OneInputStreamOperatorTestHarness<String, RowData> harness =
openHarness(conf, writeClient)) {
+ harness.processElement(new StreamRecord<>("ignored"));
+ assertTrue(harness.getOutput().isEmpty());
+ harness.snapshot(1, 1);
+ harness.notifyOfCompletedCheckpoint(1);
+ }
}
- verify(writeClient).close();
+ InOrder order = inOrder(cleanHandler);
+ order.verify(cleanHandler).clean();
+ order.verify(cleanHandler).startAsyncCleaning();
+ order.verify(cleanHandler).waitForCleaningFinish();
+ order.verify(cleanHandler).close();
+ verifyNoMoreInteractions(cleanHandler);
+ // The handler owns the write client and closes it itself.
+ verify(writeClient, never()).close();
}
@Test
- void testSnapshotDoesNotPropagateCleaningFailure() throws Exception {
+ void testCleaningIncludesMetadataTable() throws Exception {
Configuration conf = new Configuration();
conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, true);
+ conf.set(FlinkOptions.METADATA_ENABLED, true);
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.GLOBAL_RECORD_LEVEL_INDEX.name());
HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
- CleanFunctionV2<String> function = new CleanFunctionV2<>(conf);
- doThrow(new
RuntimeException("expected")).when(writeClient).startAsyncCleaning();
-
- try (OneInputStreamOperatorTestHarness<String, RowData> harness =
- openHarness(conf, function, writeClient)) {
- verify(writeClient, timeout(5000)).clean();
- assertTrue(waitUntil(() -> !function.isCleaning));
-
- harness.snapshot(1, 1);
+ HoodieFlinkWriteClient metadataClient = mock(HoodieFlinkWriteClient.class);
+
+ try (MockedStatic<StreamerUtil> streamerUtil =
mockStatic(StreamerUtil.class);
+ MockedConstruction<DefaultCleanHandler> handlers =
mockConstruction(DefaultCleanHandler.class)) {
+ streamerUtil.when(() ->
StreamerUtil.createMetadataWriteClient(writeClient)).thenReturn(metadataClient);
+ try (OneInputStreamOperatorTestHarness<String, RowData> harness =
openHarness(conf, writeClient)) {
+ harness.snapshot(1, 1);
+ harness.notifyOfCompletedCheckpoint(1);
+ }
- assertFalse(function.isCleaning);
- verify(writeClient).startAsyncCleaning();
+ assertEquals(2, handlers.constructed().size());
+ for (CleanHandler handler : handlers.constructed()) {
+ InOrder order = inOrder(handler);
+ order.verify(handler).clean();
+ order.verify(handler).startAsyncCleaning();
+ order.verify(handler).waitForCleaningFinish();
+ order.verify(handler).close();
+ verifyNoMoreInteractions(handler);
+ }
}
}
@@ -100,43 +115,31 @@ class TestCleanFunctionV2 {
Configuration conf = new Configuration();
conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, false);
HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
- CleanFunctionV2<String> function = new CleanFunctionV2<>(conf);
- try (OneInputStreamOperatorTestHarness<String, RowData> harness =
- openHarness(conf, function, writeClient)) {
- harness.snapshot(1, 1);
- harness.notifyOfCompletedCheckpoint(1);
- harness.processElement(new StreamRecord<>("ignored"));
-
- verify(writeClient, never()).clean();
- verify(writeClient, never()).startAsyncCleaning();
- verify(writeClient, never()).waitForCleaningFinish();
- assertTrue(harness.getOutput().isEmpty());
+ try (MockedStatic<TableServiceHandlerFactory> factory =
mockStatic(TableServiceHandlerFactory.class)) {
+ try (OneInputStreamOperatorTestHarness<String, RowData> harness =
openHarness(conf, writeClient)) {
+ harness.snapshot(1, 1);
+ harness.notifyOfCompletedCheckpoint(1);
+ harness.processElement(new StreamRecord<>("ignored"));
+ assertTrue(harness.getOutput().isEmpty());
+ }
+ factory.verifyNoInteractions();
}
+ verifyNoInteractions(writeClient);
}
private OneInputStreamOperatorTestHarness<String, RowData> openHarness(
Configuration conf,
- CleanFunctionV2<String> function,
HoodieFlinkWriteClient writeClient) throws Exception {
OneInputStreamOperatorTestHarness<String, RowData> harness =
- new OneInputStreamOperatorTestHarness<>(new
ProcessOperator<>(function), 1, 1, 0);
+ new OneInputStreamOperatorTestHarness<>(new ProcessOperator<>(new
CleanFunctionV2<String>(conf)), 1, 1, 0);
try (MockedStatic<FlinkWriteClients> writeClients =
mockStatic(FlinkWriteClients.class)) {
writeClients.when(() -> FlinkWriteClients.createWriteClient(
eq(conf), any())).thenReturn(writeClient);
harness.open();
+ writeClients.verify(() -> FlinkWriteClients.createWriteClient(eq(conf),
any()),
+ times(conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED) ? 1 : 0));
}
return harness;
}
-
- private boolean waitUntil(Callable<Boolean> condition) throws Exception {
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
- while (System.nanoTime() < deadline) {
- if (condition.call()) {
- return true;
- }
- Thread.sleep(10);
- }
- return condition.call();
- }
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/clustering/TestClusteringCommitSinkV2.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/clustering/TestClusteringCommitSinkV2.java
index 1614dfdd172f..5555891afde5 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/clustering/TestClusteringCommitSinkV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/clustering/TestClusteringCommitSinkV2.java
@@ -47,15 +47,19 @@ import
org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.RowData;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.MockedStatic;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
@@ -70,6 +74,7 @@ class TestClusteringCommitSinkV2 {
private Configuration conf;
private HoodieFlinkWriteClient writeClient;
+ private HoodieFlinkWriteClient cleaningClient;
private HoodieFlinkTable table;
private HoodieTableMetaClient metaClient;
private HoodieActiveTimeline activeTimeline;
@@ -82,6 +87,7 @@ class TestClusteringCommitSinkV2 {
conf = new Configuration();
conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, false);
writeClient = mock(HoodieFlinkWriteClient.class);
+ cleaningClient = mock(HoodieFlinkWriteClient.class);
table = mock(HoodieFlinkTable.class);
metaClient = mock(HoodieTableMetaClient.class);
activeTimeline = mock(HoodieActiveTimeline.class);
@@ -170,8 +176,10 @@ class TestClusteringCommitSinkV2 {
}
}
- @Test
- void testCommitSuccessfulClusteringAndCleanInline() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testCommitSuccessfulClusteringAndCloseClientOnce(boolean asyncCleaning)
throws Exception {
+ conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, asyncCleaning);
HoodieClusteringPlan plan = planWithGroups(1);
WriteStatus status = writeStatus("partition", "new-file", 0);
stubWriteConfig();
@@ -190,7 +198,20 @@ class TestClusteringCommitSinkV2 {
verify(metaClient).reloadActiveTimeline();
verify(writeClient).completeTableService(
eq(TableServiceType.CLUSTER), any(HoodieCommitMetadata.class),
same(table), eq(INSTANT));
- verify(writeClient).clean();
+ verify(writeClient, times(asyncCleaning ? 0 : 1)).clean();
+ verify(writeClient).close();
+ verify(cleaningClient, times(asyncCleaning ? 1 : 0)).clean();
+ verify(cleaningClient, times(asyncCleaning ? 1 : 0)).close();
+ }
+
+ @Test
+ void testClosesClusteringClientWhenCleaningClientCloseFails() throws
Exception {
+ conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, true);
+ doThrow(new RuntimeException("expected")).when(cleaningClient).close();
+ OneInputStreamOperatorTestHarness<ClusteringCommitEvent, RowData> harness
= openHarness();
+ assertThrows(Exception.class, harness::close);
+ verify(cleaningClient).close();
+ verify(writeClient).close();
}
private OneInputStreamOperatorTestHarness<ClusteringCommitEvent, RowData>
openHarness() throws Exception {
@@ -199,9 +220,14 @@ class TestClusteringCommitSinkV2 {
OneInputStreamOperatorTestHarness<ClusteringCommitEvent, RowData> harness =
new OneInputStreamOperatorTestHarness<>(operator, 1, 1, 0);
try (MockedStatic<FlinkWriteClients> writeClients =
mockStatic(FlinkWriteClients.class)) {
- writeClients.when(() -> FlinkWriteClients.createWriteClient(
- eq(conf), any())).thenReturn(writeClient);
+ if (conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(eq(conf),
any())).thenReturn(cleaningClient, writeClient);
+ } else {
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(eq(conf),
any())).thenReturn(writeClient);
+ }
harness.open();
+ writeClients.verify(() -> FlinkWriteClients.createWriteClient(eq(conf),
any()),
+ times(conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED) ? 2 : 1));
}
return harness;
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/compact/TestCompactionCommitSinkV2.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/compact/TestCompactionCommitSinkV2.java
index 925e5bb2d479..e08e7ab0b23c 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/compact/TestCompactionCommitSinkV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/compact/TestCompactionCommitSinkV2.java
@@ -25,15 +25,21 @@ import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.transaction.TransactionManager;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.CompactionUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.sink.compact.CompactionCommitEvent;
+import org.apache.hudi.sink.compact.handler.CleanHandler;
+import org.apache.hudi.sink.compact.handler.CompactionCommitHandler;
+import org.apache.hudi.sink.compact.handler.TableServiceHandlerFactory;
import org.apache.hudi.table.HoodieFlinkTable;
import org.apache.hudi.table.action.compact.CompactHelpers;
import org.apache.hudi.util.CompactionUtil;
import org.apache.hudi.util.FlinkWriteClients;
+import org.apache.hudi.util.StreamerUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.operators.ProcessOperator;
@@ -42,19 +48,24 @@ import
org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.RowData;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.MockedStatic;
import java.util.Arrays;
import java.util.List;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
/** Tests {@link CompactionCommitSinkV2} with Flink's operator test harness. */
@@ -64,6 +75,7 @@ class TestCompactionCommitSinkV2 {
private Configuration conf;
private HoodieFlinkWriteClient writeClient;
+ private HoodieFlinkWriteClient cleaningClient;
private HoodieFlinkTable table;
private HoodieTableMetaClient metaClient;
private TransactionManager transactionManager;
@@ -74,6 +86,7 @@ class TestCompactionCommitSinkV2 {
conf = new Configuration();
conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, false);
writeClient = mock(HoodieFlinkWriteClient.class);
+ cleaningClient = mock(HoodieFlinkWriteClient.class);
table = mock(HoodieFlinkTable.class);
metaClient = mock(HoodieTableMetaClient.class);
transactionManager = mock(TransactionManager.class);
@@ -153,9 +166,11 @@ class TestCompactionCommitSinkV2 {
}
}
- @Test
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
@SuppressWarnings("unchecked")
- void testCommitSuccessfulCompactionAndCleanInline() throws Exception {
+ void testCommitSuccessfulCompactionAndClean(boolean asyncCleaning) throws
Exception {
+ conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, asyncCleaning);
WriteStatus status = writeStatus("partition", "file-1", 0);
HoodieCommitMetadata metadata = new HoodieCommitMetadata();
CompactHelpers compactHelpers = mock(CompactHelpers.class);
@@ -175,7 +190,82 @@ class TestCompactionCommitSinkV2 {
}
verify(writeClient).completeCompaction(same(metadata), same(table),
eq(INSTANT));
- verify(writeClient).clean();
+ verify(writeClient, times(asyncCleaning ? 0 : 1)).clean();
+ verify(writeClient).close();
+ verify(cleaningClient, times(asyncCleaning ? 1 : 0)).close();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ @SuppressWarnings("unchecked")
+ void testMetadataCompactionUsesSeparateHandler(boolean logCompaction) throws
Exception {
+ conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
+ conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, true);
+ conf.set(FlinkOptions.METADATA_ENABLED, true);
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.GLOBAL_RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.METADATA_COMPACTION_ASYNC_ENABLED, true);
+ HoodieFlinkWriteClient metadataClient = mock(HoodieFlinkWriteClient.class);
+ HoodieFlinkTable metadataTable = mock(HoodieFlinkTable.class);
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ when(metadataClient.getHoodieTable()).thenReturn(metadataTable);
+ when(metadataTable.getMetaClient()).thenReturn(metadataMetaClient);
+ stubWriteConfig();
+ HoodieWriteConfig writeConfig = writeClient.getConfig();
+ when(metadataClient.getConfig()).thenReturn(writeConfig);
+ CompactHelpers compactHelpers = mock(CompactHelpers.class);
+ HoodieCommitMetadata metadata = new HoodieCommitMetadata();
+ when(compactHelpers.createCompactionMetadata(same(metadataTable),
eq(INSTANT), any(), eq("{}"),
+ eq(logCompaction ? WriteOperationType.LOG_COMPACT :
WriteOperationType.COMPACT))).thenReturn(metadata);
+
+ HoodieCompactionPlan dataPlan = planWithOperations(2);
+ HoodieCompactionPlan metadataPlan = planWithOperations(1);
+ try (MockedStatic<StreamerUtil> streamerUtil =
mockStatic(StreamerUtil.class);
+ MockedStatic<CompactionUtils> compactionUtils =
mockStatic(CompactionUtils.class);
+ MockedStatic<CompactHelpers> helpersFactory =
mockStatic(CompactHelpers.class)) {
+ streamerUtil.when(() ->
StreamerUtil.createMetadataWriteClient(writeClient)).thenReturn(metadataClient);
+
helpersFactory.when(CompactHelpers::getInstance).thenReturn(compactHelpers);
+ compactionUtils.when(() -> CompactionUtils.getCompactionPlan(metaClient,
INSTANT)).thenReturn(dataPlan);
+ if (logCompaction) {
+ compactionUtils.when(() ->
CompactionUtils.getLogCompactionPlan(metadataMetaClient,
INSTANT)).thenReturn(metadataPlan);
+ } else {
+ compactionUtils.when(() ->
CompactionUtils.getCompactionPlan(metadataMetaClient,
INSTANT)).thenReturn(metadataPlan);
+ }
+ try (OneInputStreamOperatorTestHarness<CompactionCommitEvent, RowData>
harness = openHarness()) {
+ // Identical instant and file IDs must not mix the data and metadata
buffers.
+ process(harness, successEvent("file-1", new WriteStatus()));
+ process(harness, new CompactionCommitEvent(INSTANT, "file-1",
+ Arrays.asList(writeStatus("partition", "file-1", 0)), 0, true,
logCompaction));
+ }
+ }
+
+ if (logCompaction) {
+ verify(metadataClient).completeLogCompaction(same(metadata),
same(metadataTable), eq(INSTANT));
+ verify(metadataClient, never()).completeCompaction(any(), any(), any());
+ } else {
+ verify(metadataClient).completeCompaction(same(metadata),
same(metadataTable), eq(INSTANT));
+ verify(metadataClient, never()).completeLogCompaction(any(), any(),
any());
+ }
+ verify(writeClient, never()).completeCompaction(any(), any(), any());
+ verify(metadataClient).close();
+ verify(writeClient).close();
+ verifyNoInteractions(cleaningClient);
+ }
+
+ @Test
+ void testClosesCleanHandlerWhenCommitHandlerCloseFails() throws Exception {
+ conf.set(FlinkOptions.CLEAN_ASYNC_ENABLED, true);
+ CleanHandler cleanHandler = mock(CleanHandler.class);
+ CompactionCommitHandler handler = mock(CompactionCommitHandler.class);
+ doThrow(new RuntimeException("expected")).when(handler).close();
+ try (MockedStatic<TableServiceHandlerFactory> factory =
mockStatic(TableServiceHandlerFactory.class)) {
+ factory.when(() ->
TableServiceHandlerFactory.createCompactionCommitHandler(eq(conf),
any())).thenReturn(handler);
+ factory.when(() -> TableServiceHandlerFactory.createCleanHandler(conf,
cleaningClient)).thenReturn(cleanHandler);
+ OneInputStreamOperatorTestHarness<CompactionCommitEvent, RowData>
harness = openHarness();
+ assertThrows(Exception.class, harness::close);
+ }
+ verify(handler).registerMetrics(any());
+ verify(handler).close();
+ verify(cleanHandler).close();
}
private OneInputStreamOperatorTestHarness<CompactionCommitEvent, RowData>
openHarness() throws Exception {
@@ -184,9 +274,15 @@ class TestCompactionCommitSinkV2 {
OneInputStreamOperatorTestHarness<CompactionCommitEvent, RowData> harness =
new OneInputStreamOperatorTestHarness<>(operator, 1, 1, 0);
try (MockedStatic<FlinkWriteClients> writeClients =
mockStatic(FlinkWriteClients.class)) {
- writeClients.when(() -> FlinkWriteClients.createWriteClient(
- eq(conf), any())).thenReturn(writeClient);
+ if (conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(eq(conf),
any())).thenReturn(cleaningClient, writeClient);
+ } else {
+ writeClients.when(() -> FlinkWriteClients.createWriteClient(eq(conf),
any())).thenReturn(writeClient);
+ }
harness.open();
+ if (!conf.get(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
+ writeClients.verify(() ->
FlinkWriteClients.createWriteClient(eq(conf), any()));
+ }
}
return harness;
}
