This is an automated email from the ASF dual-hosted git repository.
voonhous 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 59e62dd488cd fix: Improve error message for conflict resolution
(#18119)
59e62dd488cd is described below
commit 59e62dd488cd964c3125e9cb96d0089f9148c0be
Author: Vinish Reddy <[email protected]>
AuthorDate: Wed Jul 15 12:12:54 2026 +0530
fix: Improve error message for conflict resolution (#18119)
* Improve error message for conflict resolution
* Fix build
* Address comments
---
...urrentFileWritesConflictResolutionStrategy.java | 62 +++++++-
...urrentFileWritesConflictResolutionStrategy.java | 171 +++++++++++++++++++++
.../hudi/common/model/WriteOperationType.java | 11 ++
3 files changed, 242 insertions(+), 2 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 0d6327272a19..92c6f6f66ba5 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
@@ -200,8 +200,66 @@ public class
SimpleConcurrentFileWritesConflictResolutionStrategy
return thisOperation.getCommitMetadataOption();
}
// just abort the current write if conflicts are found (failed for
rollback conflicts).
- throw new HoodieWriteConflictException(new
ConcurrentModificationException("Cannot resolve conflicts for overlapping
writes between first operation = " + thisOperation
- + ", second operation = " + otherOperation));
+ throw new HoodieWriteConflictException(new
ConcurrentModificationException(buildConflictErrorMessage(thisOperation,
otherOperation)));
+ }
+
+ /**
+ * Builds a detailed error message for write conflicts based on the
operation types involved.
+ */
+ private String buildConflictErrorMessage(ConcurrentOperation thisOperation,
ConcurrentOperation otherOperation) {
+ boolean thisIsTableService =
WriteOperationType.isTableService(thisOperation.getOperationType());
+ boolean otherIsTableService =
WriteOperationType.isTableService(otherOperation.getOperationType());
+ String thisOperationDescription =
formatOperationDescription(thisOperation);
+ String otherOperationDescription =
formatOperationDescription(otherOperation);
+ // If either operation is a table service, provide specific retry guidance
+ if (thisIsTableService || otherIsTableService) {
+ ConcurrentOperation tableServiceOperation = thisIsTableService ?
thisOperation : otherOperation;
+ String tableServiceDescription = thisIsTableService ?
thisOperationDescription : otherOperationDescription;
+ String regularOperationDescription = thisIsTableService ?
otherOperationDescription : thisOperationDescription;
+ String serviceType =
getTableServiceDisplayName(tableServiceOperation.getOperationType());
+ return String.format(
+ "Cannot resolve conflicts for overlapping writes. %s is currently
running and has overlapping file groups with %s. "
+ + "Please retry the write operation after the %s completes.",
+ tableServiceDescription, regularOperationDescription,
serviceType.toLowerCase()
+ );
+ }
+ // For regular write operations conflicting with each other
+ return String.format(
+ "Cannot resolve conflicts for overlapping writes. %s has overlapping
file groups with %s.",
+ thisOperationDescription, otherOperationDescription
+ );
+ }
+
+ /**
+ * Formats a description of an operation including its type, instant, and
state.
+ */
+ private String formatOperationDescription(ConcurrentOperation operation) {
+ String operationName =
WriteOperationType.isTableService(operation.getOperationType())
+ ? "Table " + getTableServiceDisplayName(operation.getOperationType())
+ : operation.getOperationType().value() + " operation";
+
+ return String.format("%s (instant: %s, state: %s)",
+ operationName,
+ operation.getInstantTimestamp(),
+ operation.getInstantActionState());
+ }
+
+ /**
+ * Returns a user-friendly display name for table service operations.
+ */
+ private String getTableServiceDisplayName(WriteOperationType operationType) {
+ switch (operationType) {
+ case COMPACT:
+ return "Compaction";
+ case CLUSTER:
+ return "Clustering";
+ case LOG_COMPACT:
+ return "Log Compaction";
+ case INDEX:
+ return "Indexing";
+ default:
+ return operationType.value();
+ }
}
@Override
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 d68d6e50f1d1..04fecce34e51 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
@@ -514,4 +514,175 @@ public class
TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
}
}
}
+
+ @Test
+ public void testErrorMessageForConflictWithCompaction() throws Exception {
+ initMetaClient(true, HoodieTableType.MERGE_ON_READ);
+ createCommit(WriteClientTestUtils.createNewInstantTime(), metaClient);
+ HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
+ Option<HoodieInstant> lastSuccessfulInstant =
timeline.getCommitsTimeline().filterCompletedInstants().lastInstant();
+
+ // writer 1 starts
+ String currentWriterInstant = WriteClientTestUtils.createNewInstantTime();
+ createInflightCommit(currentWriterInstant, metaClient);
+
+ // compaction gets scheduled and runs
+ String compactionInstant = WriteClientTestUtils.createNewInstantTime();
+ createCompactionRequested(compactionInstant, metaClient);
+
+ Option<HoodieInstant> currentInstant =
Option.of(INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
+ SimpleConcurrentFileWritesConflictResolutionStrategy strategy = new
SimpleConcurrentFileWritesConflictResolutionStrategy();
+ HoodieCommitMetadata currentMetadata =
createCommitMetadata(currentWriterInstant);
+ metaClient.reloadActiveTimeline();
+
+ List<HoodieInstant> candidateInstants =
strategy.getCandidateInstants(metaClient, currentInstant.get(),
lastSuccessfulInstant)
+ .collect(Collectors.toList());
+ Assertions.assertEquals(1, candidateInstants.size());
+
+ ConcurrentOperation thatCompactionOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
+ ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
+ Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatCompactionOperation));
+
+ HoodieWriteConflictException exception =
Assertions.assertThrows(HoodieWriteConflictException.class,
+ () -> strategy.resolveConflict(null, thisCommitOperation,
thatCompactionOperation));
+
+ String errorMessage = exception.getMessage();
+ Assertions.assertTrue(errorMessage.contains("Table Compaction"),
+ "Error message should mention 'Table Compaction', but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains("is currently running"),
+ "Error message should contain 'is currently running', but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains("Please retry the write
operation after the compaction completes"),
+ "Error message should contain retry guidance, but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains(compactionInstant),
+ "Error message should contain compaction instant time, but was: " +
errorMessage);
+ }
+
+ @Test
+ public void testErrorMessageForConflictWithClustering() throws Exception {
+ initMetaClient();
+ createCommit(WriteClientTestUtils.createNewInstantTime(), metaClient);
+ HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
+ Option<HoodieInstant> lastSuccessfulInstant =
timeline.getCommitsTimeline().filterCompletedInstants().lastInstant();
+
+ // writer 1 starts
+ String currentWriterInstant = WriteClientTestUtils.createNewInstantTime();
+ createInflightCommit(currentWriterInstant, metaClient);
+
+ // clustering gets scheduled
+ String clusteringInstant = WriteClientTestUtils.createNewInstantTime();
+ createClusterRequested(clusteringInstant, metaClient);
+
+ Option<HoodieInstant> currentInstant =
Option.of(INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
+ SimpleConcurrentFileWritesConflictResolutionStrategy strategy = new
SimpleConcurrentFileWritesConflictResolutionStrategy();
+ HoodieCommitMetadata currentMetadata =
createCommitMetadata(currentWriterInstant);
+ metaClient.reloadActiveTimeline();
+
+ List<HoodieInstant> candidateInstants =
strategy.getCandidateInstants(metaClient, currentInstant.get(),
lastSuccessfulInstant)
+ .collect(Collectors.toList());
+ Assertions.assertEquals(1, candidateInstants.size());
+
+ ConcurrentOperation thatClusteringOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
+ ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
+ Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatClusteringOperation));
+
+ HoodieWriteConflictException exception =
Assertions.assertThrows(HoodieWriteConflictException.class,
+ () -> strategy.resolveConflict(null, thisCommitOperation,
thatClusteringOperation));
+
+ String errorMessage = exception.getMessage();
+ Assertions.assertTrue(errorMessage.contains("Table Clustering"),
+ "Error message should mention 'Table Clustering', but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains("is currently running"),
+ "Error message should contain 'is currently running', but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains("Please retry the write
operation after the clustering completes"),
+ "Error message should contain retry guidance, but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains(clusteringInstant),
+ "Error message should contain clustering instant time, but was: " +
errorMessage);
+ }
+
+ @Test
+ public void testErrorMessageForConflictBetweenRegularWrites() throws
Exception {
+ initMetaClient();
+ createCommit(WriteClientTestUtils.createNewInstantTime(), metaClient);
+ HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
+ Option<HoodieInstant> lastSuccessfulInstant =
timeline.getCommitsTimeline().filterCompletedInstants().lastInstant();
+
+ // writer 1 starts
+ String currentWriterInstant = WriteClientTestUtils.createNewInstantTime();
+ createInflightCommit(currentWriterInstant, metaClient);
+
+ // writer 2 starts and finishes
+ String writer2Instant = WriteClientTestUtils.createNewInstantTime();
+ createCommit(writer2Instant, metaClient);
+
+ Option<HoodieInstant> currentInstant =
Option.of(INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
+ SimpleConcurrentFileWritesConflictResolutionStrategy strategy = new
SimpleConcurrentFileWritesConflictResolutionStrategy();
+ HoodieCommitMetadata currentMetadata =
createCommitMetadata(currentWriterInstant);
+ metaClient.reloadActiveTimeline();
+
+ 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));
+
+ HoodieWriteConflictException exception =
Assertions.assertThrows(HoodieWriteConflictException.class,
+ () -> strategy.resolveConflict(null, thisCommitOperation,
thatCommitOperation));
+
+ String errorMessage = exception.getMessage();
+ Assertions.assertTrue(errorMessage.contains("Cannot resolve conflicts for
overlapping writes"),
+ "Error message should mention overlapping writes, but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains("has overlapping file groups"),
+ "Error message should contain 'has overlapping file groups', but was:
" + errorMessage);
+ Assertions.assertTrue(errorMessage.contains(currentWriterInstant),
+ "Error message should contain current writer instant time, but was: "
+ errorMessage);
+ Assertions.assertTrue(errorMessage.contains(writer2Instant),
+ "Error message should contain other writer instant time, but was: " +
errorMessage);
+ // Should NOT contain table service specific messaging
+ Assertions.assertFalse(errorMessage.contains("Table Compaction"),
+ "Error message should not mention table services for regular writes,
but was: " + errorMessage);
+ Assertions.assertFalse(errorMessage.contains("Please retry"),
+ "Error message should not contain retry guidance for regular writes,
but was: " + errorMessage);
+ }
+
+ @Test
+ public void testErrorMessageForConflictWithCompletedClustering() throws
Exception {
+ initMetaClient();
+ createCommit(WriteClientTestUtils.createNewInstantTime(), metaClient);
+ HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
+ Option<HoodieInstant> lastSuccessfulInstant =
timeline.getCommitsTimeline().filterCompletedInstants().lastInstant();
+
+ // writer 1 starts
+ String currentWriterInstant = WriteClientTestUtils.createNewInstantTime();
+ createInflightCommit(currentWriterInstant, metaClient);
+
+ // clustering completes
+ String clusteringInstant = WriteClientTestUtils.createNewInstantTime();
+ createCluster(clusteringInstant, WriteOperationType.CLUSTER, metaClient);
+
+ Option<HoodieInstant> currentInstant =
Option.of(INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
+ SimpleConcurrentFileWritesConflictResolutionStrategy strategy = new
SimpleConcurrentFileWritesConflictResolutionStrategy();
+ HoodieCommitMetadata currentMetadata =
createCommitMetadata(currentWriterInstant);
+ metaClient.reloadActiveTimeline();
+
+ List<HoodieInstant> candidateInstants =
strategy.getCandidateInstants(metaClient, currentInstant.get(),
lastSuccessfulInstant)
+ .collect(Collectors.toList());
+ Assertions.assertEquals(1, candidateInstants.size());
+
+ ConcurrentOperation thatClusteringOperation = new
ConcurrentOperation(candidateInstants.get(0), metaClient);
+ ConcurrentOperation thisCommitOperation = new
ConcurrentOperation(currentInstant.get(), currentMetadata);
+ Assertions.assertTrue(strategy.hasConflict(thisCommitOperation,
thatClusteringOperation));
+
+ HoodieWriteConflictException exception =
Assertions.assertThrows(HoodieWriteConflictException.class,
+ () -> strategy.resolveConflict(null, thisCommitOperation,
thatClusteringOperation));
+
+ String errorMessage = exception.getMessage();
+ Assertions.assertTrue(errorMessage.contains("Table Clustering"),
+ "Error message should mention 'Table Clustering', but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains(clusteringInstant),
+ "Error message should contain clustering instant time, but was: " +
errorMessage);
+ Assertions.assertTrue(errorMessage.contains("COMPLETED"),
+ "Error message should indicate the state of the clustering operation,
but was: " + errorMessage);
+ }
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/WriteOperationType.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/WriteOperationType.java
index e794aad60b60..db288339a1b0 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/model/WriteOperationType.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/model/WriteOperationType.java
@@ -188,6 +188,17 @@ public enum WriteOperationType {
return operationType == COMPACT || operationType == CLUSTER;
}
+ /**
+ * Checks if the given operation type is a table service operation.
+ * Table service operations include compaction, clustering, log compaction,
and indexing.
+ */
+ public static boolean isTableService(WriteOperationType operationType) {
+ return operationType == COMPACT
+ || operationType == CLUSTER
+ || operationType == LOG_COMPACT
+ || operationType == INDEX;
+ }
+
/**
* @return true if streaming writes to metadata table is supported for a
given {@link WriteOperationType}. false otherwise.
*/
