[HUDI-1575] Early Conflict Detection For Multi-writer

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java

@@ -89,6 +89,15 @@ public void startServer() throws IOException {
           .markerParallelism(writeConfig.getMarkersDeleteParallelism());
     }
 
+    if (writeConfig.isEarlyConflictDetectionEnable()) {
+      timelineServiceConfBuilder.earlyConflictDetectionEnable(true)
+          .earlyConflictDetectionStrategy(writeConfig.getEarlyConflictDetectionStrategyClassName())
+          .earlyConflictDetectionCheckCommitConflict(writeConfig.earlyConflictDetectionCheckCommitConflict())
+          .asyncConflictDetectorBatchIntervalMs(writeConfig.getAsyncConflictDetectorBatchIntervalMs())
+          .asyncConflictDetectorBatchPeriodMs(writeConfig.getAsyncConflictDetectorPeriodMs())
+          .earlyConflictDetectionMaxAllowableHeartbeatIntervalInMs(writeConfig.getHoodieClientHeartbeatIntervalInMs());
+    }
+
     server = new TimelineService(context, hadoopConf.newCopy(), timelineServiceConfBuilder.build(),
         FSUtils.getFs(basePath, hadoopConf.newCopy()), viewManager);
     serverPort = server.startService();

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HeartbeatUtils.java

@@ -18,18 +18,21 @@
 
 package org.apache.hudi.client.heartbeat;
 
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.table.HoodieTable;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
 import java.io.IOException;
 
+import static org.apache.hudi.common.heartbeat.HoodieHeartbeatUtils.getLastHeartbeatTime;
+
 /**
  * Helper class to delete heartbeat for completed or failed instants with expired heartbeats.
  */
@@ -89,7 +92,7 @@ public static void abortIfHeartbeatExpired(String instantTime, HoodieTable table
     try {
       if (config.getFailedWritesCleanPolicy().isLazy() && heartbeatClient.isHeartbeatExpired(instantTime)) {
         throw new HoodieException("Heartbeat for instant " + instantTime + " has expired, last heartbeat "
-            + HoodieHeartbeatClient.getLastHeartbeatTime(table.getMetaClient().getFs(), config.getBasePath(), instantTime));
+            + getLastHeartbeatTime(table.getMetaClient().getFs(), config.getBasePath(), instantTime));
       }
     } catch (IOException io) {
       throw new HoodieException("Unable to read heartbeat", io);

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java

@@ -18,17 +18,19 @@
 
 package org.apache.hudi.client.heartbeat;
 
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieHeartbeatException;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
 import javax.annotation.concurrent.NotThreadSafe;
+
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.Serializable;
@@ -37,9 +39,11 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.heartbeat.HoodieHeartbeatUtils.getLastHeartbeatTime;
 
 /**
  * This class creates heartbeat for hudi client. This heartbeat is used to ascertain whether the running job is or not.
@@ -205,16 +209,6 @@ public void stop() throws HoodieException {
     instantToHeartbeatMap.values().forEach(heartbeat -> stop(heartbeat.getInstantTime()));
   }
 
-  public static Long getLastHeartbeatTime(FileSystem fs, String basePath, String instantTime) throws IOException {
-    Path heartbeatFilePath = new Path(HoodieTableMetaClient.getHeartbeatFolderPath(basePath) + Path.SEPARATOR + instantTime);
-    if (fs.exists(heartbeatFilePath)) {
-      return fs.getFileStatus(heartbeatFilePath).getModificationTime();
-    } else {
-      // NOTE : This can happen when a writer is upgraded to use lazy cleaning and the last write had failed
-      return 0L;
-    }
-  }
-
   public static Boolean heartbeatExists(FileSystem fs, String basePath, String instantTime) throws IOException {
     Path heartbeatFilePath = new Path(HoodieTableMetaClient.getHeartbeatFolderPath(basePath) + Path.SEPARATOR + instantTime);
     if (fs.exists(heartbeatFilePath)) {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java

@@ -0,0 +1,93 @@
+/*
+ * 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.client.transaction;
+
+import org.apache.hudi.client.transaction.lock.LockManager;
+import org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider;
+import org.apache.hudi.common.config.LockConfiguration;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieNotSupportedException;
+
+import org.apache.hadoop.fs.FileSystem;
+
+import static org.apache.hudi.common.util.StringUtils.EMPTY_STRING;
+
+/**
+ * This class allows clients to start and end transactions for creating direct marker, used by
+ * `SimpleTransactionDirectMarkerBasedDetectionStrategy`, when early conflict
+ * detection is enabled.  Anything done between a start and end transaction is guaranteed to be
+ * atomic.
+ */
+public class DirectMarkerTransactionManager extends TransactionManager {
+  private final String filePath;
+
+  public DirectMarkerTransactionManager(HoodieWriteConfig config, FileSystem fs, String partitionPath, String fileId) {
+    super(new LockManager(config, fs, createUpdatedLockProps(config, partitionPath, fileId)),
+        config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl());
+    this.filePath = partitionPath + "/" + fileId;
+  }
+
+  public void beginTransaction(String newTxnOwnerInstantTime) {
+    if (isOptimisticConcurrencyControlEnabled) {
+      LOG.info("Transaction starting for " + newTxnOwnerInstantTime + " and " + filePath);
+      lockManager.lock();
+
+      reset(currentTxnOwnerInstant, Option.of(getInstant(newTxnOwnerInstantTime)), Option.empty());
+      LOG.info("Transaction started for " + newTxnOwnerInstantTime + " and " + filePath);
+    }
+  }
+
+  public void endTransaction(String currentTxnOwnerInstantTime) {
+    if (isOptimisticConcurrencyControlEnabled) {
+      LOG.info("Transaction ending with transaction owner " + currentTxnOwnerInstantTime
+          + " for " + filePath);
+      if (reset(Option.of(getInstant(currentTxnOwnerInstantTime)), Option.empty(), Option.empty())) {
+        lockManager.unlock();
+        LOG.info("Transaction ended with transaction owner " + currentTxnOwnerInstantTime
+            + " for " + filePath);
+      }
+    }
+  }
+
+  /**
+   * Rebuilds lock related configs. Only support ZK related lock for now.
+   *
+   * @param writeConfig   Hudi write configs.
+   * @param partitionPath Relative partition path.
+   * @param fileId        File ID.
+   * @return Updated lock related configs.
+   */
+  private static TypedProperties createUpdatedLockProps(
+      HoodieWriteConfig writeConfig, String partitionPath, String fileId) {
+    if (!ZookeeperBasedLockProvider.class.getName().equals(writeConfig.getLockProviderClass())) {
+      throw new HoodieNotSupportedException("Only Support ZK-based lock for DirectMarkerTransactionManager now.");
+    }
+    TypedProperties props = new TypedProperties(writeConfig.getProps());
+    props.setProperty(LockConfiguration.ZK_LOCK_KEY_PROP_KEY, partitionPath + "/" + fileId);
+    return props;
+  }
+
+  private HoodieInstant getInstant(String instantTime) {
+    return new HoodieInstant(HoodieInstant.State.INFLIGHT, EMPTY_STRING, instantTime);
+  }
+}

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/TransactionManager.java

@@ -18,11 +18,12 @@
 
 package org.apache.hudi.client.transaction;
 
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hudi.client.transaction.lock.LockManager;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
+
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
@@ -34,15 +35,20 @@
  */
 public class TransactionManager implements Serializable {
 
-  private static final Logger LOG = LogManager.getLogger(TransactionManager.class);
-  private final LockManager lockManager;
-  private final boolean isOptimisticConcurrencyControlEnabled;
-  private Option<HoodieInstant> currentTxnOwnerInstant = Option.empty();
+  protected static final Logger LOG = LogManager.getLogger(TransactionManager.class);
+  protected final LockManager lockManager;
+  protected final boolean isOptimisticConcurrencyControlEnabled;
+  protected Option<HoodieInstant> currentTxnOwnerInstant = Option.empty();
   private Option<HoodieInstant> lastCompletedTxnOwnerInstant = Option.empty();
 
   public TransactionManager(HoodieWriteConfig config, FileSystem fs) {
-    this.lockManager = new LockManager(config, fs);
-    this.isOptimisticConcurrencyControlEnabled = config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl();
+    this(new LockManager(config, fs),
+        config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl());
+  }
+
+  protected TransactionManager(LockManager lockManager, boolean isOptimisticConcurrencyControlEnabled) {
+    this.lockManager = lockManager;
+    this.isOptimisticConcurrencyControlEnabled = isOptimisticConcurrencyControlEnabled;
   }
 
   public void beginTransaction(Option<HoodieInstant> newTxnOwnerInstant,
@@ -67,9 +73,9 @@ public void endTransaction(Option<HoodieInstant> currentTxnOwnerInstant) {
     }
   }
 
-  private synchronized boolean reset(Option<HoodieInstant> callerInstant,
-                                  Option<HoodieInstant> newTxnOwnerInstant,
-                                  Option<HoodieInstant> lastCompletedTxnOwnerInstant) {
+  protected synchronized boolean reset(Option<HoodieInstant> callerInstant,
+                                       Option<HoodieInstant> newTxnOwnerInstant,
+                                       Option<HoodieInstant> lastCompletedTxnOwnerInstant) {
     if (!this.currentTxnOwnerInstant.isPresent() || this.currentTxnOwnerInstant.get().equals(callerInstant.get())) {
       this.currentTxnOwnerInstant = newTxnOwnerInstant;
       this.lastCompletedTxnOwnerInstant = lastCompletedTxnOwnerInstant;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java

@@ -21,6 +21,7 @@
 import org.apache.hudi.client.transaction.lock.metrics.HoodieLockMetrics;
 import org.apache.hudi.common.config.LockConfiguration;
 import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.lock.LockProvider;
 import org.apache.hudi.common.util.ReflectionUtils;
 import org.apache.hudi.config.HoodieLockConfig;
@@ -52,9 +53,13 @@ public class LockManager implements Serializable, AutoCloseable {
   private volatile LockProvider lockProvider;
 
   public LockManager(HoodieWriteConfig writeConfig, FileSystem fs) {
+    this(writeConfig, fs, writeConfig.getProps());
+  }
+
+  public LockManager(HoodieWriteConfig writeConfig, FileSystem fs, TypedProperties lockProps) {
     this.writeConfig = writeConfig;
     this.hadoopConf = new SerializableConfiguration(fs.getConf());
-    this.lockConfiguration = new LockConfiguration(writeConfig.getProps());
+    this.lockConfiguration = new LockConfiguration(lockProps);
     maxRetries = lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY,
         Integer.parseInt(HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.defaultValue()));
     maxWaitTimeInMs = lockConfiguration.getConfig().getLong(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY,