[HUDI-2285] Metadata Table synchronous design

hudi-cli/src/main/java/org/apache/hudi/cli/HoodieCLI.java

@@ -42,7 +42,7 @@ public class HoodieCLI {
   public static CLIState state = CLIState.INIT;
   public static String basePath;
   protected static HoodieTableMetaClient tableMetadata;
-  public static HoodieTableMetaClient syncTableMetadata;
+  public static HoodieTableMetaClient metaClient;
   public static TimelineLayoutVersion layoutVersion;
   private static TempViewProvider tempViewProvider;
 

hudi-cli/src/main/java/org/apache/hudi/cli/HoodiePrompt.java

@@ -40,7 +40,7 @@ public String getPrompt() {
         case TABLE:
           return "hudi:" + tableName + "->";
         case SYNC:
-          return "hudi:" + tableName + " <==> " + HoodieCLI.syncTableMetadata.getTableConfig().getTableName() + "->";
+          return "hudi:" + tableName + " <==> " + HoodieCLI.metaClient.getTableConfig().getTableName() + "->";
         default:
           return "hudi:" + tableName + "->";
       }

hudi-cli/src/main/java/org/apache/hudi/cli/commands/CommitsCommand.java

@@ -442,10 +442,10 @@ public String compareCommits(@CliOption(key = {"path"}, help = "Path of the tabl
 
   @CliCommand(value = "commits sync", help = "Compare commits with another Hoodie table")
   public String syncCommits(@CliOption(key = {"path"}, help = "Path of the table to compare to") final String path) {
-    HoodieCLI.syncTableMetadata = HoodieTableMetaClient.builder().setConf(HoodieCLI.conf).setBasePath(path).build();
+    HoodieCLI.metaClient = HoodieTableMetaClient.builder().setConf(HoodieCLI.conf).setBasePath(path).build();
     HoodieCLI.state = HoodieCLI.CLIState.SYNC;
     return "Load sync state between " + HoodieCLI.getTableMetaClient().getTableConfig().getTableName() + " and "
-        + HoodieCLI.syncTableMetadata.getTableConfig().getTableName();
+        + HoodieCLI.metaClient.getTableConfig().getTableName();
   }
 
   /*

hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieSyncCommand.java

@@ -56,10 +56,10 @@ public String validateSync(
       @CliOption(key = {"hivePass"}, mandatory = true, unspecifiedDefaultValue = "",
           help = "hive password to connect to") final String hivePass)
       throws Exception {
-    if (HoodieCLI.syncTableMetadata == null) {
+    if (HoodieCLI.metaClient == null) {
       throw new HoodieException("Sync validate request target table not null.");
     }
-    HoodieTableMetaClient target = HoodieCLI.syncTableMetadata;
+    HoodieTableMetaClient target = HoodieCLI.metaClient;
     HoodieTimeline targetTimeline = target.getActiveTimeline().getCommitsTimeline();
     HoodieTableMetaClient source = HoodieCLI.getTableMetaClient();
     HoodieTimeline sourceTimeline = source.getActiveTimeline().getCommitsTimeline();

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java

@@ -60,6 +60,7 @@
 import org.apache.hudi.exception.HoodieRollbackException;
 import org.apache.hudi.exception.HoodieSavepointException;
 import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
 import org.apache.hudi.metrics.HoodieMetrics;
 import org.apache.hudi.table.BulkInsertPartitioner;
 import org.apache.hudi.table.HoodieTable;
@@ -187,6 +188,7 @@ public boolean commitStats(String instantTime, List<HoodieWriteStat> stats, Opti
         lastCompletedTxnAndMetadata.isPresent() ? Option.of(lastCompletedTxnAndMetadata.get().getLeft()) : Option.empty());
     try {
       preCommit(instantTime, metadata);
+      table.getMetadataWriter().ifPresent(w -> ((HoodieTableMetadataWriter)w).update(metadata, instantTime));
       commit(table, commitActionType, instantTime, metadata, stats);
       postCommit(table, metadata, instantTime, extraMetadata);
       LOG.info("Committed " + instantTime);
@@ -241,10 +243,6 @@ protected void preCommit(String instantTime, HoodieCommitMetadata metadata) {
     // TODO : Conflict resolution is not supported for Flink & Java engines
   }
 
-  protected void syncTableMetadata() {
-    // no-op
-  }
-
   /**
    * Filter out HoodieRecords that already exists in the output folder. This is useful in deduplication.
    *
@@ -399,14 +397,6 @@ protected void preWrite(String instantTime, WriteOperationType writeOperationTyp
       HoodieTableMetaClient metaClient) {
     setOperationType(writeOperationType);
     this.lastCompletedTxnAndMetadata = TransactionUtils.getLastCompletedTxnInstantAndMetadata(metaClient);
-    this.txnManager.beginTransaction(Option.of(new HoodieInstant(State.INFLIGHT, metaClient.getCommitActionType(), instantTime)), lastCompletedTxnAndMetadata
-        .isPresent()
-        ? Option.of(lastCompletedTxnAndMetadata.get().getLeft()) : Option.empty());
-    try {
-      syncTableMetadata();
-    } finally {
-      this.txnManager.endTransaction();
-    }
     this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this);
   }
 
@@ -435,7 +425,6 @@ protected void postCommit(HoodieTable<T, I, K, O> table, HoodieCommitMetadata me
       HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(config, table);
       archiveLog.archiveIfRequired(context);
       autoCleanOnCommit();
-      syncTableMetadata();
     } catch (IOException ioe) {
       throw new HoodieIOException(ioe.getMessage(), ioe);
     } finally {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java

@@ -58,7 +58,7 @@ public static Option<HoodieCommitMetadata> resolveWriteConflictIfAny(final Hoodi
     if (config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()) {
       ConflictResolutionStrategy resolutionStrategy = config.getWriteConflictResolutionStrategy();
       Stream<HoodieInstant> instantStream = resolutionStrategy.getCandidateInstants(table.getActiveTimeline(), currentTxnOwnerInstant.get(), lastCompletedTxnOwnerInstant);
-      final ConcurrentOperation thisOperation = new ConcurrentOperation(currentTxnOwnerInstant.get(), thisCommitMetadata.get());
+      final ConcurrentOperation thisOperation = new ConcurrentOperation(currentTxnOwnerInstant.get(), thisCommitMetadata.orElse(new HoodieCommitMetadata()));
       instantStream.forEach(instant -> {
         try {
           ConcurrentOperation otherOperation = new ConcurrentOperation(instant, table.getMetaClient());

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLockConfig.java

@@ -209,6 +209,11 @@ public HoodieLockConfig.Builder withLockProvider(Class<? extends LockProvider> l
       return this;
     }
 
+    public HoodieLockConfig.Builder withLockProviderClass(String lockProviderClassName) {
+      lockConfig.setValue(LOCK_PROVIDER_CLASS_PROP, lockProviderClassName);
+      return this;
+    }
+
     public HoodieLockConfig.Builder withHiveDatabaseName(String databaseName) {
       lockConfig.setValue(HIVE_DATABASE_NAME_PROP, databaseName);
       return this;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -1220,10 +1220,6 @@ public boolean useFileListingMetadata() {
     return metadataConfig.useFileListingMetadata();
   }
 
-  public boolean getFileListingMetadataVerify() {
-    return metadataConfig.validateFileListingMetadata();
-  }
-
   public int getMetadataInsertParallelism() {
     return getInt(HoodieMetadataConfig.METADATA_INSERT_PARALLELISM_PROP);
   }