[HUDI-845] Added locking capability to allow multiple writers

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));
   }
 
   /**

hudi-client/hudi-client-common/pom.xml

@@ -154,6 +154,23 @@
       <scope>test</scope>
     </dependency>
 
+    <!-- Lock -->
+    <dependency>
+      <groupId>org.apache.curator</groupId>
+      <artifactId>curator-framework</artifactId>
+      <version>${zk-curator.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.curator</groupId>
+      <artifactId>curator-client</artifactId>
+      <version>${zk-curator.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.curator</groupId>
+      <artifactId>curator-recipes</artifactId>
+      <version>${zk-curator.version}</version>
+    </dependency>
+
     <!-- Test -->
     <dependency>
       <groupId>org.junit.jupiter</groupId>
@@ -195,6 +212,13 @@
       <artifactId>junit-platform-commons</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.curator</groupId>
+      <artifactId>curator-test</artifactId>
+      <version>${zk-curator.version}</version>
+      <scope>test</scope>
+    </dependency>
+
   </dependencies>
 
   <build>

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

@@ -0,0 +1,153 @@
+/*
+ * 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 java.io.IOException;
+import org.apache.hudi.avro.model.HoodieRequestedReplaceMetadata;
+import org.apache.hudi.client.utils.MetadataConversionUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieMetadataWrapper;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.CommitUtils;
+import java.util.Collections;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.hudi.common.util.Option;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+/**
+ * This class is used to hold all information used to identify how to resolve conflicts between instants.
+ * Since we interchange payload types between AVRO specific records and POJO's, this object serves as
+ * a common payload to manage these conversions.
+ */
+public class ConcurrentOperation {
+
+  private WriteOperationType operationType;
+  private final HoodieMetadataWrapper metadataWrapper;
+  private final Option<HoodieCommitMetadata> commitMetadataOption;
+  private final String actionState;
+  private final String actionType;
+  private final String instantTime;
+  private Set<String> mutatedFileIds = Collections.EMPTY_SET;
+
+  public ConcurrentOperation(HoodieInstant instant, HoodieTableMetaClient metaClient) throws IOException  {
+    this.metadataWrapper = new HoodieMetadataWrapper(MetadataConversionUtils.createMetaWrapper(instant, metaClient));
+    this.commitMetadataOption = Option.empty();
+    this.actionState = instant.getState().name();
+    this.actionType = instant.getAction();
+    this.instantTime = instant.getTimestamp();
+    init(instant);
+  }
+
+  public ConcurrentOperation(HoodieInstant instant, HoodieCommitMetadata commitMetadata) {
+    this.commitMetadataOption = Option.of(commitMetadata);
+    this.metadataWrapper = new HoodieMetadataWrapper(commitMetadata);
+    this.actionState = instant.getState().name();
+    this.actionType = instant.getAction();
+    this.instantTime = instant.getTimestamp();
+    init(instant);
+  }
+
+  public String getInstantActionState() {
+    return actionState;
+  }
+
+  public String getInstantActionType() {
+    return actionType;
+  }
+
+  public String getInstantTimestamp() {
+    return instantTime;
+  }
+
+  public WriteOperationType getOperationType() {
+    return operationType;
+  }
+
+  public Set<String> getMutatedFileIds() {
+    return mutatedFileIds;
+  }
+
+  public Option<HoodieCommitMetadata> getCommitMetadataOption() {
+    return commitMetadataOption;
+  }
+
+  private void init(HoodieInstant instant) {
+    if (this.metadataWrapper.isAvroMetadata()) {
+      switch (getInstantActionType()) {
+        case COMPACTION_ACTION:
+          this.operationType = WriteOperationType.COMPACT;
+          this.mutatedFileIds = this.metadataWrapper.getMetadataFromTimeline().getHoodieCompactionPlan().getOperations()
+              .stream()
+              .map(op -> op.getFileId())
+              .collect(Collectors.toSet());
+          break;
+        case COMMIT_ACTION:
+        case DELTA_COMMIT_ACTION:
+          this.mutatedFileIds = CommitUtils.getFileIdWithoutSuffixAndRelativePathsFromSpecificRecord(this.metadataWrapper.getMetadataFromTimeline().getHoodieCommitMetadata()
+              .getPartitionToWriteStats()).keySet();
+          this.operationType = WriteOperationType.fromValue(this.metadataWrapper.getMetadataFromTimeline().getHoodieCommitMetadata().getOperationType());
+          break;
+        case REPLACE_COMMIT_ACTION:
+          if (instant.isCompleted()) {
+            this.mutatedFileIds = CommitUtils.getFileIdWithoutSuffixAndRelativePathsFromSpecificRecord(
+                this.metadataWrapper.getMetadataFromTimeline().getHoodieReplaceCommitMetadata().getPartitionToWriteStats()).keySet();
+            this.operationType = WriteOperationType.fromValue(this.metadataWrapper.getMetadataFromTimeline().getHoodieReplaceCommitMetadata().getOperationType());
+          } else {
+            HoodieRequestedReplaceMetadata requestedReplaceMetadata = this.metadataWrapper.getMetadataFromTimeline().getHoodieRequestedReplaceMetadata();
+            this.mutatedFileIds = requestedReplaceMetadata
+                .getClusteringPlan().getInputGroups()
+                .stream()
+                .flatMap(ig -> ig.getSlices().stream())
+                .map(file -> file.getFileId())
+                .collect(Collectors.toSet());
+            this.operationType = WriteOperationType.CLUSTER;
+          }
+          break;
+        default:
+          throw new IllegalArgumentException("Unsupported Action Type " + getInstantActionType());
+      }
+    } else {
+      switch (getInstantActionType()) {
+        case COMMIT_ACTION:
+        case DELTA_COMMIT_ACTION:
+          this.mutatedFileIds = CommitUtils.getFileIdWithoutSuffixAndRelativePaths(this.metadataWrapper.getCommitMetadata().getPartitionToWriteStats()).keySet();
+          this.operationType = this.metadataWrapper.getCommitMetadata().getOperationType();
+          break;
+        default:
+          throw new IllegalArgumentException("Unsupported Action Type " + getInstantActionType());
+      }
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "{"
+        + "actionType=" + this.getInstantActionType()
+        + ", instantTime=" + this.getInstantTimestamp()
+        + ", actionState=" + this.getInstantActionState()
+        + '\'' + '}';
+  }
+}

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

@@ -0,0 +1,64 @@
+/*
+ * 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.ApiMaturityLevel;
+import org.apache.hudi.PublicAPIMethod;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieWriteConflictException;
+import org.apache.hudi.table.HoodieTable;
+
+import java.util.stream.Stream;
+
+/**
+ * Strategy interface for conflict resolution with multiple writers.
+ * Users can provide pluggable implementations for different kinds of strategies to resolve conflicts when multiple
+ * writers are mutating the hoodie table.
+ */
+public interface ConflictResolutionStrategy {
+
+  /**
+   * Stream of instants to check conflicts against.
+   * @return
+   */
+  Stream<HoodieInstant> getCandidateInstants(HoodieActiveTimeline activeTimeline, HoodieInstant currentInstant, Option<HoodieInstant> lastSuccessfulInstant);
+
+  /**
+   * Implementations of this method will determine whether a conflict exists between 2 commits.
+   * @param thisOperation
+   * @param otherOperation
+   * @return
+   */
+  @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+  boolean hasConflict(ConcurrentOperation thisOperation, ConcurrentOperation otherOperation);
+
+  /**
+   * Implementations of this method will determine how to resolve a conflict between 2 commits.
+   * @param thisOperation
+   * @param otherOperation
+   * @return
+   */
+  @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+  Option<HoodieCommitMetadata> resolveConflict(HoodieTable table,
+      ConcurrentOperation thisOperation, ConcurrentOperation otherOperation) throws HoodieWriteConflictException;
+
+}

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

@@ -0,0 +1,105 @@
+/*
+ * 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.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieWriteConflictException;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.util.ConcurrentModificationException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+
+/**
+ * This class is a basic implementation of a conflict resolution strategy for concurrent writes {@link ConflictResolutionStrategy}.
+ */
+public class SimpleConcurrentFileWritesConflictResolutionStrategy
+    implements ConflictResolutionStrategy {
+
+  private static final Logger LOG = LogManager.getLogger(SimpleConcurrentFileWritesConflictResolutionStrategy.class);
+
+  @Override
+  public Stream<HoodieInstant> getCandidateInstants(HoodieActiveTimeline activeTimeline, HoodieInstant currentInstant,
+                                                 Option<HoodieInstant> lastSuccessfulInstant) {
+
+    // To find which instants are conflicting, we apply the following logic
+    // 1. Get completed instants timeline only for commits that have happened since the last successful write.
+    // 2. Get any scheduled or completed compaction or clustering operations that have started and/or finished
+    // after the current instant. We need to check for write conflicts since they may have mutated the same files
+    // that are being newly created by the current write.
+    Stream<HoodieInstant> completedCommitsInstantStream = activeTimeline
+        .getCommitsTimeline()
+        .filterCompletedInstants()
+        .findInstantsAfter(lastSuccessfulInstant.isPresent() ? lastSuccessfulInstant.get().getTimestamp() : HoodieTimeline.INIT_INSTANT_TS)
+        .getInstants();
+
+    Stream<HoodieInstant> compactionAndClusteringPendingTimeline = activeTimeline
+        .getTimelineOfActions(CollectionUtils.createSet(REPLACE_COMMIT_ACTION, COMPACTION_ACTION))
+        .findInstantsAfter(currentInstant.getTimestamp())
+        .filterInflightsAndRequested()
+        .getInstants();
+    return Stream.concat(completedCommitsInstantStream, compactionAndClusteringPendingTimeline);
+  }
+
+  @Override
+  public boolean hasConflict(ConcurrentOperation thisOperation, ConcurrentOperation otherOperation) {
+    // TODO : UUID's can clash even for insert/insert, handle that case.
+    Set<String> fileIdsSetForFirstInstant = thisOperation.getMutatedFileIds();
+    Set<String> fileIdsSetForSecondInstant = otherOperation.getMutatedFileIds();
+    Set<String> intersection = new HashSet<>(fileIdsSetForFirstInstant);
+    intersection.retainAll(fileIdsSetForSecondInstant);
+    if (!intersection.isEmpty()) {
+      LOG.info("Found conflicting writes between first operation = " + thisOperation
+          + ", second operation = " + otherOperation + " , intersecting file ids " + intersection);
+      return true;
+    }
+    return false;
+  }
+
+  @Override
+  public Option<HoodieCommitMetadata> resolveConflict(HoodieTable table,
+      ConcurrentOperation thisOperation, ConcurrentOperation otherOperation) {
+    // A completed COMPACTION action eventually shows up as a COMMIT action on the timeline.
+    // We need to ensure we handle this during conflict resolution and not treat the commit from a
+    // compaction operation as a regular commit. Regular commits & deltacommits are candidates for conflict.
+    // Since the REPLACE action with CLUSTER operation does not support concurrent updates, we have
+    // to consider it as conflict if we see overlapping file ids. Once concurrent updates are
+    // supported for CLUSTER (https://issues.apache.org/jira/browse/HUDI-1042),
+    // add that to the below check so that concurrent updates do not conflict.
+    if (otherOperation.getOperationType() == WriteOperationType.COMPACT
+        && HoodieTimeline.compareTimestamps(otherOperation.getInstantTimestamp(), HoodieTimeline.LESSER_THAN, thisOperation.getInstantTimestamp())) {
+      return thisOperation.getCommitMetadataOption();
+    }
+    // just abort the current write if conflicts are found
+    throw new HoodieWriteConflictException(new ConcurrentModificationException("Cannot resolve conflicts for overlapping writes"));
+  }
+
+}