diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/testutils/HoodieTestCommitUtilities.java b/hudi-cli/src/test/java/org/apache/hudi/cli/testutils/HoodieTestCommitUtilities.java
index 8829c858051fb..13f8a74faa5d9 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/testutils/HoodieTestCommitUtilities.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/testutils/HoodieTestCommitUtilities.java
@@ -19,8 +19,8 @@
package org.apache.hudi.cli.testutils;
import org.apache.hudi.avro.model.HoodieWriteStat;
+import org.apache.hudi.client.utils.MetadataConversionUtils;
import org.apache.hudi.common.model.HoodieCommitMetadata;
-import org.apache.hudi.table.HoodieTimelineArchiveLog;
import java.util.LinkedHashMap;
import java.util.List;
@@ -36,7 +36,7 @@ public class HoodieTestCommitUtilities {
*/
public static org.apache.hudi.avro.model.HoodieCommitMetadata convertAndOrderCommitMetadata(
HoodieCommitMetadata hoodieCommitMetadata) {
- return orderCommitMetadata(HoodieTimelineArchiveLog.convertCommitMetadata(hoodieCommitMetadata));
+ return orderCommitMetadata(MetadataConversionUtils.convertCommitMetadata(hoodieCommitMetadata));
}
/**
diff --git a/hudi-client/hudi-client-common/pom.xml b/hudi-client/hudi-client-common/pom.xml
index 676251e0670de..afca6a62d0d93 100644
--- a/hudi-client/hudi-client-common/pom.xml
+++ b/hudi-client/hudi-client-common/pom.xml
@@ -154,6 +154,23 @@
test
+
+
+ org.apache.curator
+ curator-framework
+ ${zk-curator.version}
+
+
+ org.apache.curator
+ curator-client
+ ${zk-curator.version}
+
+
+ org.apache.curator
+ curator-recipes
+ ${zk-curator.version}
+
+
org.junit.jupiter
@@ -195,6 +212,13 @@
junit-platform-commons
test
+
+ org.apache.curator
+ curator-test
+ ${zk-curator.version}
+ test
+
+
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
index a3ba00895e4a1..9f3b74b0c4708 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
@@ -22,6 +22,7 @@
import java.util.stream.Stream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
import org.apache.hudi.avro.model.HoodieRestoreMetadata;
@@ -31,15 +32,20 @@
import org.apache.hudi.callback.util.HoodieCommitCallbackFactory;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
import org.apache.hudi.client.heartbeat.HeartbeatUtils;
+import org.apache.hudi.client.transaction.TransactionManager;
+import org.apache.hudi.client.utils.TransactionUtils;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.TableServiceType;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieInstant.State;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.CleanerUtils;
import org.apache.hudi.common.util.ClusteringUtils;
@@ -101,6 +107,8 @@ public abstract class AbstractHoodieWriteClient>> lastCompletedTxnAndMetadata = Option.empty();
/**
* Create a write client, with new hudi index.
@@ -124,6 +132,7 @@ public AbstractHoodieWriteClient(HoodieEngineContext context, HoodieWriteConfig
super(context, writeConfig, timelineService);
this.metrics = new HoodieMetrics(config, config.getTableName());
this.index = createIndex(writeConfig);
+ this.txnManager = new TransactionManager(config, fs);
}
protected abstract HoodieIndex createIndex(HoodieWriteConfig writeConfig);
@@ -163,26 +172,28 @@ public boolean commitStats(String instantTime, List stats, Opti
public boolean commitStats(String instantTime, List stats, Option