This is an automated email from the ASF dual-hosted git repository.
vinoth pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 450a3c27ce4 [HUDI-9438] Fix conflict handling for compaction instants
for v8 tables (#13347)
450a3c27ce4 is described below
commit 450a3c27ce4a31524ac6d6a971a39c4ad35490ac
Author: Tim Brown <[email protected]>
AuthorDate: Fri May 30 09:45:46 2025 -0500
[HUDI-9438] Fix conflict handling for compaction instants for v8 tables
(#13347)
* fix conflict handling for compaction given completion time changes
* consolidate tests
* split handling into two methods for ease of reading and debugging
* extract common parts of the code
---
...urrentFileWritesConflictResolutionStrategy.java | 71 +++++++++++--
...urrentFileWritesConflictResolutionStrategy.java | 111 ++++++++++++++++-----
...itesConflictResolutionStrategyWithMORTable.java | 90 -----------------
.../hudi/client/TestHoodieClientMultiWriter.java | 21 +---
4 files changed, 155 insertions(+), 138 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
index 8cbc141e0e8..a3686288ba0 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
@@ -19,8 +19,10 @@
package org.apache.hudi.client.transaction;
import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
@@ -36,6 +38,7 @@ import org.slf4j.LoggerFactory;
import java.util.ConcurrentModificationException;
import java.util.HashSet;
import java.util.Set;
+import java.util.function.Predicate;
import java.util.stream.Stream;
import static
org.apache.hudi.common.table.timeline.InstantComparison.GREATER_THAN;
@@ -53,26 +56,80 @@ public class
SimpleConcurrentFileWritesConflictResolutionStrategy
@Override
public Stream<HoodieInstant> getCandidateInstants(HoodieTableMetaClient
metaClient, HoodieInstant currentInstant,
Option<HoodieInstant>
lastSuccessfulInstant) {
+ if
(metaClient.getTableConfig().getTableVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT))
{
+ return getCandidateInstantsV8AndAbove(metaClient, currentInstant,
lastSuccessfulInstant);
+ } else {
+ return getCandidateInstantsPreV8(metaClient, currentInstant,
lastSuccessfulInstant);
+ }
+ }
+
+ /**
+ * To find which instants are conflicting for table versions 8 and above, we
apply the following logic:
+ * <ul>
+ * <li>Get completed instants timeline only for commits that have happened
since the last successful write.</li>
+ * <li>Get any completed replace commit that happened since the last
successful write and any pending replace commit.</li>
+ * </ul>
+ * @param metaClient table meta client
+ * @param currentInstant the instant for the write this client is attempting
to commit
+ * @param lastSuccessfulInstant the last successful write before this commit
started
+ * @return a stream of instants that are candidates for conflict resolution
+ */
+ private Stream<HoodieInstant>
getCandidateInstantsV8AndAbove(HoodieTableMetaClient metaClient, HoodieInstant
currentInstant,
+
Option<HoodieInstant> lastSuccessfulInstant) {
HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
- // 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 that have started and/or
finished after the current instant.
- // 3. Get any completed replace commit that happened since the last
successful write and any pending replace commit.
- // We need to check for write conflicts since they may have mutated the
same files that are being newly created by the current write.
+ boolean isMoRTable = metaClient.getTableType() ==
HoodieTableType.MERGE_ON_READ;
Stream<HoodieInstant> completedCommitsInstantStream = activeTimeline
.getCommitsTimeline()
.filterCompletedInstants()
+ .filter(instant -> !isMoRTable ||
!instant.getAction().equals(HoodieTimeline.COMMIT_ACTION))
.findInstantsAfter(lastSuccessfulInstant.isPresent() ?
lastSuccessfulInstant.get().requestedTime() : HoodieTimeline.INIT_INSTANT_TS)
.getInstantsAsStream();
+ Stream<HoodieInstant> clusteringAndReplaceCommitInstants = activeTimeline
+ .filterPendingReplaceOrClusteringTimeline()
+ .filter(instant ->
isClusteringOrRecentlyRequestedInstant(activeTimeline, metaClient,
currentInstant).test(instant))
+ .getInstantsAsStream();
+
+ return Stream.concat(completedCommitsInstantStream,
clusteringAndReplaceCommitInstants);
+ }
+
+ /**
+ * To find which instants are conflicting for table versions below 8, we
apply the following logic:
+ * <ul>
+ * <li>Get completed instants timeline only for commits that have happened
since the last successful write.</li>
+ * <li>Get any scheduled or completed compaction that have started and/or
finished after the current instant.</li>
+ * <li>Get any completed replace commit that happened since the last
successful write and any pending replace commit.</li>
+ * </ul>
+ * @param metaClient table meta client
+ * @param currentInstant the instant for the write this client is attempting
to commit
+ * @param lastSuccessfulInstant the last successful write before this commit
started
+ * @return a stream of instants that are candidates for conflict resolution
+ */
+ private Stream<HoodieInstant>
getCandidateInstantsPreV8(HoodieTableMetaClient metaClient, HoodieInstant
currentInstant,
+
Option<HoodieInstant> lastSuccessfulInstant) {
+ HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
+ Stream<HoodieInstant> completedCommitsInstantStream =
getCommitsCompletedSinceLastCommit(lastSuccessfulInstant, activeTimeline);
+
Stream<HoodieInstant> compactionAndClusteringPendingTimeline =
activeTimeline
.filterPendingReplaceClusteringAndCompactionTimeline()
- .filter(instant -> ClusteringUtils.isClusteringInstant(activeTimeline,
instant, metaClient.getInstantGenerator())
- || (!HoodieTimeline.CLUSTERING_ACTION.equals(instant.getAction())
&& compareTimestamps(instant.requestedTime(), GREATER_THAN,
currentInstant.requestedTime())))
+ .filter(instant ->
isClusteringOrRecentlyRequestedInstant(activeTimeline, metaClient,
currentInstant).test(instant))
.getInstantsAsStream();
return Stream.concat(completedCommitsInstantStream,
compactionAndClusteringPendingTimeline);
}
+ private static Stream<HoodieInstant>
getCommitsCompletedSinceLastCommit(Option<HoodieInstant> lastSuccessfulInstant,
HoodieActiveTimeline activeTimeline) {
+ return activeTimeline
+ .getCommitsTimeline()
+ .filterCompletedInstants()
+
.findInstantsAfter(lastSuccessfulInstant.map(HoodieInstant::requestedTime).orElse(HoodieTimeline.INIT_INSTANT_TS))
+ .getInstantsAsStream();
+ }
+
+ private Predicate<HoodieInstant>
isClusteringOrRecentlyRequestedInstant(HoodieActiveTimeline activeTimeline,
HoodieTableMetaClient metaClient, HoodieInstant currentInstant) {
+ return instant -> ClusteringUtils.isClusteringInstant(activeTimeline,
instant, metaClient.getInstantGenerator())
+ || (!HoodieTimeline.CLUSTERING_ACTION.equals(instant.getAction()) &&
compareTimestamps(instant.requestedTime(), GREATER_THAN,
currentInstant.requestedTime()));
+ }
+
@Override
public boolean hasConflict(ConcurrentOperation thisOperation,
ConcurrentOperation otherOperation) {
// TODO : UUID's can clash even for insert/insert, handle that case.
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java
index a749fce98f4..443cd02ff02 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java
@@ -20,6 +20,7 @@ package org.apache.hudi.client.transaction;
import org.apache.hudi.client.utils.TransactionUtils;
import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
@@ -31,10 +32,10 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieWriteConflictException;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
-import java.io.IOException;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
@@ -60,13 +61,9 @@ import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR
public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends
HoodieCommonTestHarness {
- @BeforeEach
- public void init() throws IOException {
- initMetaClient();
- }
-
@Test
public void testNoConcurrentWrites() throws Exception {
+ initMetaClient();
String newInstantTime = HoodieTestTable.makeNewCommitTime();
createCommit(newInstantTime, metaClient);
// consider commits before this are all successful
@@ -82,6 +79,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWrites() throws Exception {
+ initMetaClient();
String newInstantTime = HoodieTestTable.makeNewCommitTime();
createCommit(newInstantTime, metaClient);
// consider commits before this are all successful
@@ -99,6 +97,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithInterleavingSuccessfulCommit() throws
Exception {
+ initMetaClient();
createCommit(metaClient.createNewInstantTime(), metaClient);
HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
// consider commits before this are all successful
@@ -126,6 +125,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithReplaceInflightCommit() throws Exception
{
+ initMetaClient();
String currentWriterInstant = metaClient.createNewInstantTime();
createInflightCommit(currentWriterInstant, metaClient);
Option<HoodieInstant> currentInstant =
Option.of(INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
@@ -152,6 +152,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithClusteringInflightCommit() throws
Exception {
+ initMetaClient();
// writer 1 starts
String currentWriterInstant = metaClient.createNewInstantTime();
createInflightCommit(currentWriterInstant, metaClient);
@@ -179,6 +180,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithLegacyClusteringInflightCommit() throws
Exception {
+ initMetaClient();
String clusteringInstantTime = metaClient.createNewInstantTime();
// create a replace commit with a clustering operation to mimic a commit
written by a v6 writer
HoodieTestTable.of(metaClient).addRequestedReplace(clusteringInstantTime,
Option.of(buildRequestedReplaceMetadata("file-1", WriteOperationType.CLUSTER)));
@@ -205,8 +207,10 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
Assertions.assertThrows(HoodieWriteConflictException.class, () ->
strategy.resolveConflict(null, thisCommitOperation, thatCommitOperation));
}
- @Test
- public void testConcurrentWritesWithInterleavingScheduledCompaction() throws
Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testConcurrentWritesWithInterleavingScheduledCompaction(boolean
preTableVersion8) throws Exception {
+ initMetaClient(preTableVersion8, HoodieTableType.MERGE_ON_READ);
createCommit(metaClient.createNewInstantTime(), metaClient);
HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
// consider commits before this are all successful
@@ -224,16 +228,23 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
metaClient.reloadActiveTimeline();
List<HoodieInstant> candidateInstants =
strategy.getCandidateInstants(metaClient, currentInstant.get(),
lastSuccessfulInstant).collect(
Collectors.toList());
- // writer 1 conflicts with scheduled compaction plan 1
- Assertions.assertEquals(1, candidateInstants.size());
- ConcurrentOperation thatCommitOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
- ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
- Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatCommitOperation));
- Assertions.assertThrows(HoodieWriteConflictException.class, () ->
strategy.resolveConflict(null, thisCommitOperation, thatCommitOperation));
+ if (preTableVersion8) {
+ // writer 1 conflicts with scheduled compaction plan 1
+ Assertions.assertEquals(1, candidateInstants.size());
+ ConcurrentOperation thatCommitOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
+ ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
+ Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatCommitOperation));
+ Assertions.assertThrows(HoodieWriteConflictException.class, () ->
strategy.resolveConflict(null, thisCommitOperation, thatCommitOperation));
+ } else {
+ // writer will not have conflicts with compaction since ordering is now
based on completion time to avoid these conflicts
+ Assertions.assertTrue(candidateInstants.isEmpty());
+ }
}
- @Test
- public void testConcurrentWritesWithInterleavingSuccessfulCompaction()
throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testConcurrentWritesWithInterleavingSuccessfulCompaction(boolean
preTableVersion8) throws Exception {
+ initMetaClient(preTableVersion8, HoodieTableType.MERGE_ON_READ);
createCommit(metaClient.createNewInstantTime(), metaClient);
HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
// consider commits before this are all successful
@@ -251,12 +262,60 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
metaClient.reloadActiveTimeline();
List<HoodieInstant> candidateInstants =
strategy.getCandidateInstants(metaClient, currentInstant.get(),
lastSuccessfulInstant).collect(
Collectors.toList());
- // writer 1 conflicts with compaction 1
- Assertions.assertEquals(1, candidateInstants.size());
- ConcurrentOperation thatCommitOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
- ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
- Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatCommitOperation));
- Assertions.assertThrows(HoodieWriteConflictException.class, () ->
strategy.resolveConflict(null, thisCommitOperation, thatCommitOperation));
+ if (preTableVersion8) {
+ // writer 1 conflicts with compaction 1
+ Assertions.assertEquals(1, candidateInstants.size());
+ ConcurrentOperation thatCommitOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
+ ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
+ Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatCommitOperation));
+ Assertions.assertThrows(HoodieWriteConflictException.class, () ->
strategy.resolveConflict(null, thisCommitOperation, thatCommitOperation));
+ } else {
+ // writer will not have conflicts with compaction since ordering is now
based on completion time to avoid these conflicts
+ Assertions.assertTrue(candidateInstants.isEmpty());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testConcurrentWritesWithInterleavingInflightCompaction(boolean
preTableVersion8) throws Exception {
+ initMetaClient(preTableVersion8, HoodieTableType.MERGE_ON_READ);
+ createCommit(metaClient.createNewInstantTime(), metaClient);
+ HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
+ // Consider commits before this are all successful.
+ Option<HoodieInstant> lastSuccessfulInstant =
timeline.getCommitsTimeline().filterCompletedInstants().lastInstant();
+
+ // Writer 1 starts.
+ String currentWriterInstant = metaClient.createNewInstantTime();
+ createInflightCommit(currentWriterInstant, metaClient);
+
+ // Compaction 1 gets scheduled and becomes inflight.
+ String newInstantTime = metaClient.createNewInstantTime();
+ createPendingCompaction(newInstantTime, metaClient);
+
+ // Writer 1 tries to commit.
+ Option<HoodieInstant> currentInstant = Option.of(
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.DELTA_COMMIT_ACTION, currentWriterInstant));
+ HoodieCommitMetadata currentMetadata =
createCommitMetadata(currentWriterInstant);
+ metaClient.reloadActiveTimeline();
+
+ // Do conflict resolution.
+ SimpleConcurrentFileWritesConflictResolutionStrategy strategy =
+ new SimpleConcurrentFileWritesConflictResolutionStrategy();
+ List<HoodieInstant> candidateInstants = strategy.getCandidateInstants(
+ metaClient, currentInstant.get(),
lastSuccessfulInstant).collect(Collectors.toList());
+
+ if (preTableVersion8) {
+ Assertions.assertEquals(1, candidateInstants.size());
+ ConcurrentOperation thatCommitOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
+ ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
+ Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatCommitOperation));
+ Assertions.assertThrows(
+ HoodieWriteConflictException.class,
+ () -> strategy.resolveConflict(null, thisCommitOperation,
thatCommitOperation));
+ } else {
+ // Writer will not have conflicts with compaction since ordering is now
based on completion time to avoid these conflicts.
+ Assertions.assertTrue(candidateInstants.isEmpty());
+ }
}
/**
@@ -264,6 +323,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
*/
@Test
public void testConcurrentWriteAndCompactionScheduledEarlier() throws
Exception {
+ initMetaClient(false, HoodieTableType.MERGE_ON_READ);
createCommit(metaClient.createNewInstantTime(), metaClient);
// compaction 1 gets scheduled
String newInstantTime = metaClient.createNewInstantTime();
@@ -286,6 +346,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithInterleavingScheduledCluster() throws
Exception {
+ initMetaClient(false, HoodieTableType.MERGE_ON_READ);
createCommit(metaClient.createNewInstantTime(), metaClient);
HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
// consider commits before this are all successful
@@ -313,6 +374,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithInterleavingSuccessfulCluster() throws
Exception {
+ initMetaClient();
createCommit(metaClient.createNewInstantTime(), metaClient);
HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
// consider commits before this are all successful
@@ -340,6 +402,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithInterleavingSuccessfulReplace() throws
Exception {
+ initMetaClient();
createCommit(metaClient.createNewInstantTime(), metaClient);
HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
// consider commits before this are all successful
@@ -367,6 +430,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithPendingInsertOverwriteReplace() throws
Exception {
+ initMetaClient();
createCommit(metaClient.createNewInstantTime(), metaClient);
HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
// consider commits before this are all successful
@@ -396,6 +460,7 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
// try to simulate HUDI-3355
@Test
public void testConcurrentWritesWithPendingInstants() throws Exception {
+ initMetaClient(false, HoodieTableType.MERGE_ON_READ);
// step1: create a pending replace/commit/compact instant: C1,C11,C12
String newInstantTimeC1 = metaClient.createNewInstantTime();
createPendingCluster(newInstantTimeC1, WriteOperationType.CLUSTER,
metaClient);
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategyWithMORTable.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategyWithMORTable.java
deleted file mode 100644
index c94d3009cd9..00000000000
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategyWithMORTable.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.HoodieTableType;
-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.testutils.HoodieCommonTestHarness;
-import org.apache.hudi.common.util.Option;
-import org.apache.hudi.exception.HoodieWriteConflictException;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.stream.Collectors;
-
-import static
org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCommit;
-import static
org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCommitMetadata;
-import static
org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createInflightCommit;
-import static
org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createPendingCompaction;
-import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
-
-public class
TestSimpleConcurrentFileWritesConflictResolutionStrategyWithMORTable extends
HoodieCommonTestHarness {
- @Override
- protected HoodieTableType getTableType() {
- return HoodieTableType.MERGE_ON_READ;
- }
-
- @BeforeEach
- public void init() throws IOException {
- initMetaClient();
- }
-
- @Test
- public void testConcurrentWritesWithInterleavingInflightCompaction() throws
Exception {
- createCommit(metaClient.createNewInstantTime(), metaClient);
- HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
- // Consider commits before this are all successful.
- Option<HoodieInstant> lastSuccessfulInstant =
timeline.getCommitsTimeline().filterCompletedInstants().lastInstant();
-
- // Writer 1 starts.
- String currentWriterInstant = metaClient.createNewInstantTime();
- createInflightCommit(currentWriterInstant, metaClient);
-
- // Compaction 1 gets scheduled and becomes inflight.
- String newInstantTime = metaClient.createNewInstantTime();
- createPendingCompaction(newInstantTime, metaClient);
-
- // Writer 1 tries to commit.
- Option<HoodieInstant> currentInstant = Option.of(
- INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.DELTA_COMMIT_ACTION, currentWriterInstant));
- HoodieCommitMetadata currentMetadata =
createCommitMetadata(currentWriterInstant);
- metaClient.reloadActiveTimeline();
-
- // Do conflict resolution.
- SimpleConcurrentFileWritesConflictResolutionStrategy strategy =
- new SimpleConcurrentFileWritesConflictResolutionStrategy();
- List<HoodieInstant> candidateInstants = strategy.getCandidateInstants(
- metaClient, currentInstant.get(),
lastSuccessfulInstant).collect(Collectors.toList());
- Assertions.assertEquals(1, candidateInstants.size());
- ConcurrentOperation thatCommitOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
- ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
- Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatCommitOperation));
- Assertions.assertThrows(
- HoodieWriteConflictException.class,
- () -> strategy.resolveConflict(null, thisCommitOperation,
thatCommitOperation));
- }
-}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
index 70f05439b9f..328457ac8be 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java
@@ -827,24 +827,9 @@ public class TestHoodieClientMultiWriter extends
HoodieClientTestBase {
// We want the upsert to go through only after the compaction
// and cleaning schedule completion. So, waiting on latch here.
latchCountDownAndWait(scheduleCountDownLatch, waitAndRunSecond);
- if (tableType == HoodieTableType.MERGE_ON_READ && !(resolutionStrategy
instanceof PreferWriterConflictResolutionStrategy)) {
- // HUDI-6897: Improve
SimpleConcurrentFileWritesConflictResolutionStrategy for NB-CC
- // There is no need to throw concurrent modification exception for the
simple strategy under NB-CC, because the compactor would finally resolve the
conflicts instead.
-
- // Since the concurrent modifications went in, this upsert has
- // to fail
- assertThrows(HoodieWriteConflictException.class, () -> {
- createCommitWithUpserts(cfg, client1, thirdCommitTime,
Option.of(commitTimeBetweenPrevAndNew), upsertCommitTime, numRecords);
- });
- } else {
- // We don't have the compaction for COW and so this upsert
- // has to pass
- final String newCommitTime = client1.createNewInstantTime();
- assertDoesNotThrow(() -> {
- createCommitWithUpserts(cfg, client1, thirdCommitTime,
Option.of(commitTimeBetweenPrevAndNew), newCommitTime, numRecords);
- });
- validInstants.add(newCommitTime);
- }
+ // Writes should pass since scheduled compaction does not conflict with
upsert for v8 and above
+ assertDoesNotThrow(() -> createCommitWithUpserts(cfg, client1,
thirdCommitTime, Option.of(commitTimeBetweenPrevAndNew), upsertCommitTime,
numRecords));
+ validInstants.add(upsertCommitTime);
});
Future future2 = executors.submit(() -> {
