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

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

@@ -45,6 +45,11 @@ public TransactionManager(HoodieWriteConfig config, FileSystem fs) {
     this.isOptimisticConcurrencyControlEnabled = config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl();
   }
 
+  public TransactionManager(HoodieWriteConfig config, FileSystem fs, String partitionPath, String fileId) {
+    this.lockManager = new LockManager(config, fs, partitionPath, fileId);
+    this.isOptimisticConcurrencyControlEnabled = config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl();
+  }
+
   public void beginTransaction(Option<HoodieInstant> newTxnOwnerInstant,
                                Option<HoodieInstant> lastCompletedTxnOwnerInstant) {
     if (isOptimisticConcurrencyControlEnabled) {
@@ -57,6 +62,14 @@ public void beginTransaction(Option<HoodieInstant> newTxnOwnerInstant,
     }
   }
 
+  public void beginTransaction(String partitionPath, String fileId) {
+    if (isOptimisticConcurrencyControlEnabled) {
+      LOG.info("Transaction starting for " + partitionPath + "/" + fileId);
+      lockManager.lock();
+      LOG.info("Transaction started for " + partitionPath + "/" + fileId);
+    }
+  }
+
   public void endTransaction(Option<HoodieInstant> currentTxnOwnerInstant) {
     if (isOptimisticConcurrencyControlEnabled) {
       LOG.info("Transaction ending with transaction owner " + currentTxnOwnerInstant);
@@ -67,6 +80,14 @@ public void endTransaction(Option<HoodieInstant> currentTxnOwnerInstant) {
     }
   }
 
+  public void endTransaction(String filePath) {
+    if (isOptimisticConcurrencyControlEnabled) {
+      LOG.info("Transaction ending with transaction for " + filePath);
+      lockManager.unlock();
+      LOG.info("Transaction ended with transaction for " + filePath);
+    }
+  }
+
   private synchronized boolean reset(Option<HoodieInstant> callerInstant,
                                   Option<HoodieInstant> newTxnOwnerInstant,
                                   Option<HoodieInstant> 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;
@@ -62,6 +63,33 @@ public LockManager(HoodieWriteConfig writeConfig, FileSystem fs) {
     metrics = new HoodieLockMetrics(writeConfig);
   }
 
+  /**
+   * Try to have a lock at partitionPath + fileID level for different write handler.
+   * @param writeConfig
+   * @param fs
+   * @param partitionPath
+   * @param fileId
+   */
+  public LockManager(HoodieWriteConfig writeConfig, FileSystem fs, String partitionPath, String fileId) {
+    this.writeConfig = writeConfig;
+    this.hadoopConf = new SerializableConfiguration(fs.getConf());
+    TypedProperties props = refreshLockConfig(writeConfig, partitionPath + "/" + fileId);
+    this.lockConfiguration = new LockConfiguration(props);
+    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,
+        Long.parseLong(HoodieLockConfig.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.defaultValue()));
+  }
+
+  /**
+   * rebuild lock related configs, only support ZK related lock for now.
+   */
+  private TypedProperties refreshLockConfig(HoodieWriteConfig writeConfig, String key) {
+    TypedProperties props = new TypedProperties(writeConfig.getProps());
+    props.setProperty(LockConfiguration.ZK_LOCK_KEY_PROP_KEY, key);
+    return props;
+  }
+
   public void lock() {
     if (writeConfig.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()) {
       LockProvider lockProvider = getLockProvider();

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

@@ -26,6 +26,7 @@
 import org.apache.hudi.common.config.HoodieConfig;
 import org.apache.hudi.common.lock.LockProvider;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.timeline.service.handlers.marker.AsyncTimelineMarkerEarlyConflictDetectionStrategy;
 
 import java.io.File;
 import java.io.FileReader;
@@ -189,6 +190,33 @@ public class HoodieLockConfig extends HoodieConfig {
       .withDocumentation("Lock provider class name, this should be subclass of "
           + "org.apache.hudi.client.transaction.ConflictResolutionStrategy");
 
+  // Pluggable strategies to use when early conflict detection
+  public static final ConfigProperty<String> EARLY_CONFLICT_DETECTION_STRATEGY_CLASS_NAME = ConfigProperty
+      .key(LOCK_PREFIX + "early.conflict.detection.strategy")
+      .defaultValue(AsyncTimelineMarkerEarlyConflictDetectionStrategy.class.getName())
+      .sinceVersion("0.12.0")
+      .withDocumentation("Early conflict detection class name, this should be subclass of "
+          + "org.apache.hudi.common.conflict.detection.HoodieEarlyConflictDetectionStrategy");
+
+  public static final ConfigProperty<Boolean> EARLY_CONFLICT_DETECTION_ENABLE = ConfigProperty
+      .key(LOCK_PREFIX + "early.conflict.detection.enable")
+      .defaultValue(false)
+      .sinceVersion("0.12.0")
+      .withDocumentation("Enable early conflict detection based on markers. It will try to detect writing conflict before create markers and fast fail"
+          + " which will release cluster resources as soon as possible.");
+
+  public static final ConfigProperty<Long> MARKER_CONFLICT_CHECKER_BATCH_INTERVAL = ConfigProperty
+      .key(LOCK_PREFIX + "early.conflict.async.checker.batch.interval")
+      .defaultValue(30000L)
+      .sinceVersion("0.12.0")
+      .withDocumentation("Used for timeline based marker AsyncTimelineMarkerConflictResolutionStrategy. The time to delay first async marker conflict checking.");
+
+  public static final ConfigProperty<Long> MARKER_CONFLICT_CHECKER_PERIOD = ConfigProperty
+      .key(LOCK_PREFIX + "early.conflict.async.checker.period")
+      .defaultValue(30000L)
+      .sinceVersion("0.12.0")
+      .withDocumentation("Used for timeline based marker AsyncTimelineMarkerConflictResolutionStrategy. The period between each marker conflict checking.");
+
   /** @deprecated Use {@link #WRITE_CONFLICT_RESOLUTION_STRATEGY_CLASS_NAME} and its methods instead */
   @Deprecated
   public static final String WRITE_CONFLICT_RESOLUTION_STRATEGY_CLASS_PROP = WRITE_CONFLICT_RESOLUTION_STRATEGY_CLASS_NAME.key();
@@ -311,6 +339,26 @@ public HoodieLockConfig.Builder withConflictResolutionStrategy(ConflictResolutio
       return this;
     }
 
+    public HoodieLockConfig.Builder withEarlyConflictDetectionEnable(boolean enable) {
+      lockConfig.setValue(EARLY_CONFLICT_DETECTION_ENABLE, String.valueOf(enable));
+      return this;
+    }
+
+    public HoodieLockConfig.Builder withMarkerConflictCheckerBatchInterval(long interval) {
+      lockConfig.setValue(MARKER_CONFLICT_CHECKER_BATCH_INTERVAL, String.valueOf(interval));
+      return this;
+    }
+
+    public HoodieLockConfig.Builder withMarkerConflictCheckerPeriod(long period) {
+      lockConfig.setValue(MARKER_CONFLICT_CHECKER_PERIOD, String.valueOf(period));
+      return this;
+    }
+
+    public HoodieLockConfig.Builder withEarlyConflictDetectionStrategy(String className) {
+      lockConfig.setValue(EARLY_CONFLICT_DETECTION_STRATEGY_CLASS_NAME, className);
+      return this;
+    }
+
     public HoodieLockConfig.Builder withFileSystemLockPath(String path) {
       lockConfig.setValue(FILESYSTEM_LOCK_PATH, path);
       return this;

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

@@ -30,6 +30,7 @@
 import org.apache.hudi.common.config.HoodieMetadataConfig;
 import org.apache.hudi.common.config.HoodieMetastoreConfig;
 import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.conflict.detection.HoodieEarlyConflictDetectionStrategy;
 import org.apache.hudi.common.engine.EngineType;
 import org.apache.hudi.common.fs.ConsistencyGuardConfig;
 import org.apache.hudi.common.fs.FileSystemRetryConfig;
@@ -2129,6 +2130,14 @@ public ConflictResolutionStrategy getWriteConflictResolutionStrategy() {
     return ReflectionUtils.loadClass(getString(HoodieLockConfig.WRITE_CONFLICT_RESOLUTION_STRATEGY_CLASS_NAME));
   }
 
+  public String getMarkerConflictCheckerBatchInterval() {
+    return String.valueOf(getLong(HoodieLockConfig.MARKER_CONFLICT_CHECKER_BATCH_INTERVAL));
+  }
+
+  public String getMarkerConflictCheckerPeriod() {
+    return String.valueOf(getLong(HoodieLockConfig.MARKER_CONFLICT_CHECKER_PERIOD));
+  }
+
   public Long getLockAcquireWaitTimeoutInMs() {
     return getLong(HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS);
   }
@@ -2137,6 +2146,18 @@ public WriteConcurrencyMode getWriteConcurrencyMode() {
     return WriteConcurrencyMode.fromValue(getString(WRITE_CONCURRENCY_MODE));
   }
 
+  public boolean isEarlyConflictDetectionEnable() {
+    return getBoolean(HoodieLockConfig.EARLY_CONFLICT_DETECTION_ENABLE);
+  }
+
+  public String getEarlyConflictDetectionStrategyClassName() {
+    return getString(HoodieLockConfig.EARLY_CONFLICT_DETECTION_STRATEGY_CLASS_NAME);
+  }
+
+  public HoodieEarlyConflictDetectionStrategy getEarlyConflictDetectionStrategy() {
+    return ReflectionUtils.loadClass(getString(HoodieLockConfig.EARLY_CONFLICT_DETECTION_STRATEGY_CLASS_NAME));
+  }
+
   // misc configs
   public Boolean doSkipDefaultPartitionValidation() {
     return getBoolean(SKIP_DEFAULT_PARTITION_VALIDATION);

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java

@@ -20,6 +20,9 @@
 
 import org.apache.hudi.avro.HoodieAvroUtils;
 import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.transaction.TransactionManager;
+import org.apache.hudi.common.conflict.detection.HoodieEarlyConflictDetectionStrategy;
+import org.apache.hudi.common.conflict.detection.HoodieTransactionDirectMarkerBasedEarlyConflictDetectionStrategy;
 import org.apache.hudi.common.engine.TaskContextSupplier;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.FileSlice;
@@ -28,6 +31,7 @@
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.model.IOType;
 import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.util.HoodieTimer;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.ReflectionUtils;
@@ -37,6 +41,7 @@
 import org.apache.hudi.io.storage.HoodieFileWriter;
 import org.apache.hudi.io.storage.HoodieFileWriterFactory;
 import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.marker.WriteMarkers;
 import org.apache.hudi.table.marker.WriteMarkersFactory;
 
 import org.apache.avro.Schema;
@@ -51,6 +56,8 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.apache.hudi.common.util.StringUtils.isNullOrEmpty;
 
@@ -187,8 +194,54 @@ protected Path makeNewFilePath(String partitionPath, String fileName) {
    * @param partitionPath Partition path
    */
   protected void createMarkerFile(String partitionPath, String dataFileName) {
-    WriteMarkersFactory.get(config.getMarkersType(), hoodieTable, instantTime)
-        .create(partitionPath, dataFileName, getIOType());
+    WriteMarkers writeMarkers = WriteMarkersFactory.get(config.getMarkersType(), hoodieTable, instantTime);
+    // do early conflict detection before create markers.
+    if (config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()
+        && config.isEarlyConflictDetectionEnable()) {
+      HoodieEarlyConflictDetectionStrategy earlyConflictDetectionStrategy = config.getEarlyConflictDetectionStrategy();
+      if (earlyConflictDetectionStrategy instanceof HoodieTransactionDirectMarkerBasedEarlyConflictDetectionStrategy) {
+        createMarkerWithTransaction(earlyConflictDetectionStrategy, writeMarkers, partitionPath, dataFileName);
+      } else {
+        createMarkerWithEarlyConflictDetection(earlyConflictDetectionStrategy, writeMarkers, partitionPath, dataFileName);
+      }
+    } else {
+      // create marker directly
+      writeMarkers.create(partitionPath, dataFileName, getIOType());
+    }
+  }
+
+  private Option<Path> createMarkerWithEarlyConflictDetection(HoodieEarlyConflictDetectionStrategy resolutionStrategy,
+                                                              WriteMarkers writeMarkers,
+                                                              String partitionPath,
+                                                              String dataFileName) {
+    Set<HoodieInstant> completedCommitInstants = hoodieTable.getMetaClient().getActiveTimeline()
+        .getCommitsTimeline()
+        .filterCompletedInstants()
+        .getInstants()
+        .collect(Collectors.toSet());
+
+    return writeMarkers.createWithEarlyConflictDetection(partitionPath, dataFileName, getIOType(), false, resolutionStrategy, completedCommitInstants, config, fileId);
+
+  }
+
+  private Option<Path> createMarkerWithTransaction(HoodieEarlyConflictDetectionStrategy resolutionStrategy,
+                                                   WriteMarkers writeMarkers,
+                                                   String partitionPath,
+                                                   String dataFileName) {
+    TransactionManager txnManager = new TransactionManager(config, fs, partitionPath, fileId);
+    try {
+      // Need to do transaction before create marker file when using early conflict detection
+      txnManager.beginTransaction(partitionPath, fileId);
+      return createMarkerWithEarlyConflictDetection(resolutionStrategy, writeMarkers, partitionPath, dataFileName);
+
+    } catch (Exception e) {
+      LOG.warn("Exception occurs during create marker file in early conflict detection mode.");
+      throw e;
+    } finally {
+      // End transaction after created marker file.
+      txnManager.endTransaction(partitionPath + "/" + fileId);
+      txnManager.close();
+    }
   }
 
   public Schema getWriterSchemaWithMetaFields() {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/DirectWriteMarkers.java

@@ -18,14 +18,20 @@
 
 package org.apache.hudi.table.marker;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.conflict.detection.HoodieEarlyConflictDetectionStrategy;
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.conflict.detection.HoodieDirectMarkerBasedEarlyConflictDetectionStrategy;
 import org.apache.hudi.common.model.IOType;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
 import org.apache.hudi.common.util.HoodieTimer;
 import org.apache.hudi.common.util.MarkerUtils;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.table.HoodieTable;
@@ -155,6 +161,24 @@ protected Option<Path> create(String partitionPath, String dataFileName, IOType
     return create(getMarkerPath(partitionPath, dataFileName, type), checkIfExists);
   }
 
+  @Override
+  public Option<Path> createWithEarlyConflictDetection(String partitionPath, String dataFileName, IOType type, boolean checkIfExists,
+                                                       HoodieEarlyConflictDetectionStrategy resolutionStrategy,
+                                                       Set<HoodieInstant> completedCommitInstants, HoodieWriteConfig config, String fileId) {
+    HoodieDirectMarkerBasedEarlyConflictDetectionStrategy strategy = (HoodieDirectMarkerBasedEarlyConflictDetectionStrategy) resolutionStrategy;
+    HoodieTableMetaClient metaClient =
+        HoodieTableMetaClient.builder().setConf(new Configuration()).setBasePath(config.getBasePath())
+            .setLoadActiveTimelineOnLoad(true).setConsistencyGuardConfig(config.getConsistencyGuardConfig())
+            .setLayoutVersion(Option.of(new TimelineLayoutVersion(config.getTimelineLayoutVersion())))
+            .setFileSystemRetryConfig(config.getFileSystemRetryConfig())
+            .setProperties(config.getProps()).build();
+
+    if (strategy.hasMarkerConflict(basePath, fs, partitionPath, fileId, instantTime, completedCommitInstants, metaClient)) {
+      strategy.resolveMarkerConflict(basePath, partitionPath, fileId);
+    }
+    return create(getMarkerPath(partitionPath, dataFileName, type), checkIfExists);
+  }
+
   private Option<Path> create(Path markerPath, boolean checkIfExists) {
     HoodieTimer timer = HoodieTimer.start();
     Path dirPath = markerPath.getParent();

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/SimpleDirectMarkerBasedEarlyConflictDetectionStrategy.java

@@ -0,0 +1,56 @@
+/*
+ * 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.table.marker;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.common.conflict.detection.HoodieDirectMarkerBasedEarlyConflictDetectionStrategy;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.exception.HoodieEarlyConflictDetectionException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.ConcurrentModificationException;
+import java.util.Set;
+
+/**
+ * This strategy is used for direct marker writers, trying to do early conflict detection.
+ * It will use fileSystem api like list and exist directly to check if there is any marker file conflict.
+ */
+public class SimpleDirectMarkerBasedEarlyConflictDetectionStrategy extends HoodieDirectMarkerBasedEarlyConflictDetectionStrategy {
+  private static final Logger LOG = LogManager.getLogger(SimpleDirectMarkerBasedEarlyConflictDetectionStrategy.class);
+
+  @Override
+  public boolean hasMarkerConflict(String basePath, FileSystem fs, String partitionPath, String fileId, String instantTime,
+                                   Set<HoodieInstant> completedCommitInstants, HoodieTableMetaClient metaClient) {
+    try {
+      return checkMarkerConflict(basePath, partitionPath, fileId, fs, instantTime) || checkCommitConflict(metaClient, completedCommitInstants, fileId);
+    } catch (IOException e) {
+      LOG.warn("Exception occurs during create marker file in eager conflict detection mode.");
+      throw new HoodieIOException("Exception occurs during create marker file in eager conflict detection mode.", e);
+    }
+  }
+
+  @Override
+  public void resolveMarkerConflict(String basePath, String partitionPath, String dataFileName) {
+    throw new HoodieEarlyConflictDetectionException(new ConcurrentModificationException("Early conflict detected but cannot resolve conflicts for overlapping writes"));
+  }
+}