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

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

@@ -86,7 +86,17 @@ public void startServer() throws IOException {
           .enableMarkerRequests(true)
           .markerBatchNumThreads(writeConfig.getMarkersTimelineServerBasedBatchNumThreads())
           .markerBatchIntervalMs(writeConfig.getMarkersTimelineServerBasedBatchIntervalMs())
-          .markerParallelism(writeConfig.getMarkersDeleteParallelism());
+          .markerParallelism(writeConfig.getMarkersDeleteParallelism())
+      ;
+    }
+
+    if (writeConfig.isEarlyConflictDetectionEnable()) {
+      timelineServiceConfBuilder.markerEarlyConflictDetectEnable(true)
+          .markerEarlyConflictDetectStrategy(writeConfig.getEarlyConflictDetectionStrategyClassName())
+          .markerEarlyConflictDetectCheckCommitConflict(writeConfig.earlyConflictDetectionCheckCommitConflict())
+          .markerEarlyConflictAsyncCheckerBatchInterval(writeConfig.getEarlyConflictDetectionAsyncCheckerBatchInterval())
+          .markerEarlyConflictAsyncCheckerBatchPeriod(writeConfig.getEarlyConflictDetectionAsyncCheckerPeriod())
+          .markerEarlyConflictMaxAllowableHeartbeatIntervalInMs(writeConfig.getHoodieClientHeartbeatIntervalInMs());
     }
 
     server = new TimelineService(context, hadoopConf.newCopy(), timelineServiceConfBuilder.build(),

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,15 @@ public void endTransaction(Option<HoodieInstant> currentTxnOwnerInstant) {
     }
   }
 
+  public void endTransaction(String partitionPath, String fileId) {
+    if (isOptimisticConcurrencyControlEnabled) {
+      String filePath = partitionPath + "/" + fileId;
+      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

@@ -18,16 +18,20 @@
 
 package org.apache.hudi.client.transaction.lock;
 
+import org.apache.hadoop.conf.Configuration;
 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.common.util.StringUtils;
 import org.apache.hudi.config.HoodieLockConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieLockException;
 
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.exception.HoodieNotSupportedException;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
@@ -43,23 +47,52 @@
 public class LockManager implements Serializable, AutoCloseable {
 
   private static final Logger LOG = LogManager.getLogger(LockManager.class);
-  private final HoodieWriteConfig writeConfig;
-  private final LockConfiguration lockConfiguration;
-  private final SerializableConfiguration hadoopConf;
-  private final int maxRetries;
-  private final long maxWaitTimeInMs;
+  private HoodieWriteConfig writeConfig;
+  private LockConfiguration lockConfiguration;
+  private SerializableConfiguration hadoopConf;
+  private int maxRetries;
+  private long maxWaitTimeInMs;
   private transient HoodieLockMetrics metrics;
   private volatile LockProvider lockProvider;
 
   public LockManager(HoodieWriteConfig writeConfig, FileSystem fs) {
+    init(writeConfig, fs.getConf(), writeConfig.getProps());
+  }
+
+  /**
+   * 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) {
+    TypedProperties props = refreshLockConfig(writeConfig, partitionPath + "/" + fileId);
+    init(writeConfig, fs.getConf(), props);
+  }
+
+  private void init(HoodieWriteConfig writeConfig, Configuration conf, TypedProperties lockProps) {
+    this.lockConfiguration = new LockConfiguration(lockProps);
     this.writeConfig = writeConfig;
-    this.hadoopConf = new SerializableConfiguration(fs.getConf());
-    this.lockConfiguration = new LockConfiguration(writeConfig.getProps());
-    maxRetries = lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY,
+    this.hadoopConf = new SerializableConfiguration(conf);
+    this.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,
+    this.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()));
-    metrics = new HoodieLockMetrics(writeConfig);
+    this.metrics = new HoodieLockMetrics(writeConfig);
+  }
+
+  /**
+   * rebuild lock related configs, only support ZK related lock for now.
+   */
+  private TypedProperties refreshLockConfig(HoodieWriteConfig writeConfig, String key) {
+    TypedProperties props = new TypedProperties(writeConfig.getProps());
+    String zkBasePath = props.getProperty(LockConfiguration.ZK_BASE_PATH_PROP_KEY);
+    if (StringUtils.isNullOrEmpty(zkBasePath)) {
+      throw new HoodieNotSupportedException("Only Support ZK based lock for now.");
+    }
+    props.setProperty(LockConfiguration.ZK_LOCK_KEY_PROP_KEY, key);
+    return props;
   }
 
   public void lock() {

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

@@ -69,9 +69,11 @@
 import org.apache.hudi.table.action.cluster.ClusteringPlanPartitionFilterMode;
 import org.apache.hudi.table.action.compact.CompactionTriggerStrategy;
 import org.apache.hudi.table.action.compact.strategy.CompactionStrategy;
+import org.apache.hudi.table.marker.SimpleDirectMarkerBasedEarlyConflictDetectionStrategy;
 import org.apache.hudi.table.storage.HoodieStorageLayout;
 
 import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hudi.timeline.service.handlers.marker.AsyncTimelineMarkerEarlyConflictDetectionStrategy;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 import org.apache.orc.CompressionKind;
@@ -112,6 +114,8 @@ public class HoodieWriteConfig extends HoodieConfig {
   // It is here so that both the client and deltastreamer use the same reference
   public static final String DELTASTREAMER_CHECKPOINT_KEY = "deltastreamer.checkpoint.key";
 
+  public static final String CONCURRENCY_PREFIX = "hoodie.write.concurrency.";
+
   public static final ConfigProperty<String> TBL_NAME = ConfigProperty
       .key(HoodieTableConfig.HOODIE_TABLE_NAME_KEY)
       .noDefaultValue()
@@ -512,6 +516,50 @@ public class HoodieWriteConfig extends HoodieConfig {
       .withDocumentation("When table is upgraded from pre 0.12 to 0.12, we check for \"default\" partition and fail if found one. "
           + "Users are expected to rewrite the data in those partitions. Enabling this config will bypass this validation");
 
+  // Pluggable strategies to use when early conflict detection
+  public static final ConfigProperty<String> EARLY_CONFLICT_DETECTION_STRATEGY_CLASS_NAME = ConfigProperty
+      .key(CONCURRENCY_PREFIX + "early.conflict.detection.strategy")
+      .noDefaultValue()
+      .sinceVersion("0.13.0")
+      .withInferFunction(cfg -> {
+        MarkerType markerType = MarkerType.valueOf(cfg.getString(MARKERS_TYPE).toUpperCase());
+        switch (markerType) {
+          case DIRECT:
+            return Option.of(SimpleDirectMarkerBasedEarlyConflictDetectionStrategy.class.getName());
+          case TIMELINE_SERVER_BASED:
+          default:
+            return Option.of(AsyncTimelineMarkerEarlyConflictDetectionStrategy.class.getName());
+        }
+      })
+      .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(CONCURRENCY_PREFIX + "early.conflict.detection.enable")
+      .defaultValue(false)
+      .sinceVersion("0.13.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(CONCURRENCY_PREFIX + "early.conflict.async.checker.batch.interval")
+      .defaultValue(30000L)
+      .sinceVersion("0.13.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(CONCURRENCY_PREFIX + "early.conflict.async.checker.period")
+      .defaultValue(30000L)
+      .sinceVersion("0.13.0")
+      .withDocumentation("Used for timeline based marker AsyncTimelineMarkerConflictResolutionStrategy. The period between each marker conflict checking.");
+
+  public static final ConfigProperty<Boolean> MARKER_CONFLICT_CHECK_COMMIT_CONFLICT = ConfigProperty
+      .key(CONCURRENCY_PREFIX + "early.conflict.check.commit.conflict")
+      .defaultValue(false)
+      .sinceVersion("0.13.0")
+      .withDocumentation("Enable check commit conflict or not during early conflict detect");
+
+
   private ConsistencyGuardConfig consistencyGuardConfig;
   private FileSystemRetryConfig fileSystemRetryConfig;
 
@@ -2129,6 +2177,14 @@ public ConflictResolutionStrategy getWriteConflictResolutionStrategy() {
     return ReflectionUtils.loadClass(getString(HoodieLockConfig.WRITE_CONFLICT_RESOLUTION_STRATEGY_CLASS_NAME));
   }
 
+  public Long getEarlyConflictDetectionAsyncCheckerBatchInterval() {
+    return getLong(MARKER_CONFLICT_CHECKER_BATCH_INTERVAL);
+  }
+
+  public Long getEarlyConflictDetectionAsyncCheckerPeriod() {
+    return getLong(MARKER_CONFLICT_CHECKER_PERIOD);
+  }
+
   public Long getLockAcquireWaitTimeoutInMs() {
     return getLong(HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS);
   }
@@ -2137,6 +2193,18 @@ public WriteConcurrencyMode getWriteConcurrencyMode() {
     return WriteConcurrencyMode.fromValue(getString(WRITE_CONCURRENCY_MODE));
   }
 
+  public boolean isEarlyConflictDetectionEnable() {
+    return getBoolean(EARLY_CONFLICT_DETECTION_ENABLE);
+  }
+
+  public String getEarlyConflictDetectionStrategyClassName() {
+    return getString(EARLY_CONFLICT_DETECTION_STRATEGY_CLASS_NAME);
+  }
+
+  public boolean earlyConflictDetectionCheckCommitConflict() {
+    return getBoolean(MARKER_CONFLICT_CHECK_COMMIT_CONFLICT);
+  }
+
   // misc configs
   public Boolean doSkipDefaultPartitionValidation() {
     return getBoolean(SKIP_DEFAULT_PARTITION_VALIDATION);
@@ -2649,6 +2717,31 @@ public Builder doSkipDefaultPartitionValidation(boolean skipDefaultPartitionVali
       return this;
     }
 
+    public Builder withEarlyConflictDetectionEnable(boolean enable) {
+      writeConfig.setValue(EARLY_CONFLICT_DETECTION_ENABLE, String.valueOf(enable));
+      return this;
+    }
+
+    public Builder withMarkerConflictCheckerBatchInterval(long interval) {
+      writeConfig.setValue(MARKER_CONFLICT_CHECKER_BATCH_INTERVAL, String.valueOf(interval));
+      return this;
+    }
+
+    public Builder withMarkerConflictCheckerPeriod(long period) {
+      writeConfig.setValue(MARKER_CONFLICT_CHECKER_PERIOD, String.valueOf(period));
+      return this;
+    }
+
+    public Builder withCheckCommitConflict(boolean enable) {
+      writeConfig.setValue(MARKER_CONFLICT_CHECK_COMMIT_CONFLICT, String.valueOf(enable));
+      return this;
+    }
+
+    public Builder withEarlyConflictDetectionStrategy(String className) {
+      writeConfig.setValue(EARLY_CONFLICT_DETECTION_STRATEGY_CLASS_NAME, className);
+      return this;
+    }
+
     protected void setDefaults() {
       writeConfig.setDefaultValue(MARKERS_TYPE, getDefaultMarkersType(engineType));
       // Check for mandatory properties

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

@@ -40,5 +40,5 @@ public abstract class HoodieIOHandle<T extends HoodieRecordPayload, I, K, O> {
     this.fs = getFileSystem();
   }
 
-  protected abstract FileSystem getFileSystem();
+  public abstract FileSystem getFileSystem();
 }

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

@@ -45,7 +45,7 @@ public HoodieReadHandle(HoodieWriteConfig config, HoodieTable<T, I, K, O> hoodie
   }
 
   @Override
-  protected FileSystem getFileSystem() {
+  public FileSystem getFileSystem() {
     return hoodieTable.getMetaClient().getFs();
   }
 

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

@@ -27,6 +27,7 @@
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.model.IOType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.log.HoodieLogFormat;
 import org.apache.hudi.common.util.HoodieTimer;
 import org.apache.hudi.common.util.Option;
@@ -188,7 +189,7 @@ protected Path makeNewFilePath(String partitionPath, String fileName) {
    */
   protected void createMarkerFile(String partitionPath, String dataFileName) {
     WriteMarkersFactory.get(config.getMarkersType(), hoodieTable, instantTime)
-        .create(partitionPath, dataFileName, getIOType());
+        .create(partitionPath, dataFileName, getIOType(), Option.of(this));
   }
 
   public Schema getWriterSchemaWithMetaFields() {
@@ -256,10 +257,18 @@ public String getPartitionPath() {
   public abstract IOType getIOType();
 
   @Override
-  protected FileSystem getFileSystem() {
+  public FileSystem getFileSystem() {
     return hoodieTable.getMetaClient().getFs();
   }
 
+  public HoodieWriteConfig getConfig() {
+    return this.config;
+  }
+
+  public HoodieTableMetaClient getHoodieTableMetaClient() {
+    return hoodieTable.getMetaClient();
+  }
+
   protected int getPartitionId() {
     return taskContextSupplier.getPartitionIdSupplier().get();
   }
@@ -312,6 +321,10 @@ protected HoodieLogFormat.Writer createLogWriter(String baseCommitTime, String f
     }
   }
 
+  public String getFileId() {
+    return this.fileId;
+  }
+
   private static class IgnoreRecord implements GenericRecord {
 
     @Override

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

@@ -19,13 +19,18 @@
 package org.apache.hudi.table.marker;
 
 import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.conflict.detection.HoodieDirectMarkerBasedEarlyConflictDetectionStrategy;
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.IOType;
 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.util.HoodieTimer;
 import org.apache.hudi.common.util.MarkerUtils;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
+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 +160,20 @@ 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, Set<HoodieInstant> completedCommitInstants,
+                                                       HoodieWriteConfig config, String fileId, HoodieActiveTimeline activeTimeline) {
+
+    long maxAllowableHeartbeatIntervalInMs = config.getHoodieClientHeartbeatIntervalInMs() * config.getHoodieClientHeartbeatTolerableMisses();
+
+    HoodieDirectMarkerBasedEarlyConflictDetectionStrategy strategy =
+        (HoodieDirectMarkerBasedEarlyConflictDetectionStrategy) ReflectionUtils.loadClass(config.getEarlyConflictDetectionStrategyClassName(),
+        basePath, fs, partitionPath, fileId, instantTime, activeTimeline, config, config.earlyConflictDetectionCheckCommitConflict(), maxAllowableHeartbeatIntervalInMs, completedCommitInstants);
+
+    strategy.detectAndResolveConflictIfNecessary();
+    return create(getMarkerPath(partitionPath, dataFileName, type), checkIfExists);
+  }
+
   private Option<Path> create(Path markerPath, boolean checkIfExists) {
     HoodieTimer timer = HoodieTimer.start();
     Path dirPath = markerPath.getParent();