You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "yihua (via GitHub)" <gi...@apache.org> on 2026/04/05 05:22:08 UTC
Re: [PR] [HUDI-7503] logcompaction execution should fail if another active writer is already executing the same plan [hudi]
yihua commented on code in PR #18305:
URL: https://github.com/apache/hudi/pull/18305#discussion_r3036428677
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -224,25 +224,46 @@ protected HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime, boo
+ "timestamp exists. Instant details %s", compactionInstantWithGreaterTimestamp.get()));
}
- HoodieTimeline pendingLogCompactionTimeline = table.getActiveTimeline().filterPendingLogCompactionTimeline();
InstantGenerator instantGenerator = table.getMetaClient().getInstantGenerator();
HoodieInstant inflightInstant = instantGenerator.getLogCompactionInflightInstant(logCompactionInstantTime);
- if (pendingLogCompactionTimeline.containsInstant(inflightInstant)) {
- log.info("Found Log compaction inflight file. Rolling back the commit and exiting.");
- table.rollbackInflightLogCompaction(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), txnManager);
- table.getMetaClient().reloadActiveTimeline();
- throw new HoodieException("Execution is aborted since it found an Inflight logcompaction,"
- + "log compaction plans are mutable plans, so reschedule another logcompaction.");
+ boolean isMultiWriter = config.getWriteConcurrencyMode().supportsMultiWriter();
+ if (isMultiWriter) {
+ try {
+ txnManager.beginStateChange(Option.of(inflightInstant), txnManager.getLastCompletedTransactionOwner());
+ validateHeartBeat(logCompactionInstantTime);
+ if (!table.getMetaClient().reloadActiveTimeline().filterPendingLogCompactionTimeline().containsInstant(logCompactionInstantTime)) {
+ throw new HoodieException("Requested log compaction instant " + logCompactionInstantTime
Review Comment:
🤖 nit: The nested `try-finally` for the heartbeat client is good for ensuring `heartbeatClient.stop()` is called. Could you add a comment explaining why `txnManager.endStateChange` is called outside this `try-finally` and specifically at this location, possibly before `validateHeartBeat`?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -224,25 +224,46 @@ protected HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime, boo
+ "timestamp exists. Instant details %s", compactionInstantWithGreaterTimestamp.get()));
}
- HoodieTimeline pendingLogCompactionTimeline = table.getActiveTimeline().filterPendingLogCompactionTimeline();
InstantGenerator instantGenerator = table.getMetaClient().getInstantGenerator();
HoodieInstant inflightInstant = instantGenerator.getLogCompactionInflightInstant(logCompactionInstantTime);
- if (pendingLogCompactionTimeline.containsInstant(inflightInstant)) {
- log.info("Found Log compaction inflight file. Rolling back the commit and exiting.");
- table.rollbackInflightLogCompaction(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), txnManager);
- table.getMetaClient().reloadActiveTimeline();
- throw new HoodieException("Execution is aborted since it found an Inflight logcompaction,"
- + "log compaction plans are mutable plans, so reschedule another logcompaction.");
+ boolean isMultiWriter = config.getWriteConcurrencyMode().supportsMultiWriter();
+ if (isMultiWriter) {
+ try {
+ txnManager.beginStateChange(Option.of(inflightInstant), txnManager.getLastCompletedTransactionOwner());
+ validateHeartBeat(logCompactionInstantTime);
+ if (!table.getMetaClient().reloadActiveTimeline().filterPendingLogCompactionTimeline().containsInstant(logCompactionInstantTime)) {
+ throw new HoodieException("Requested log compaction instant " + logCompactionInstantTime
+ + " is not present as pending or already completed in the active timeline.");
+ }
+ this.heartbeatClient.start(logCompactionInstantTime);
+ } finally {
+ txnManager.endStateChange(Option.of(inflightInstant));
+ }
}
- logCompactionTimer = metrics.getLogCompactionCtx();
- WriteMarkersFactory.get(config.getMarkersType(), table, logCompactionInstantTime);
- HoodieWriteMetadata<T> writeMetadata = table.logCompact(context, logCompactionInstantTime);
- HoodieWriteMetadata<T> updatedWriteMetadata = partialUpdateTableMetadata(table, writeMetadata, logCompactionInstantTime, WriteOperationType.LOG_COMPACT);
- HoodieWriteMetadata<O> logCompactionMetadata = convertToOutputMetadata(updatedWriteMetadata);
- if (shouldComplete) {
- commitLogCompaction(logCompactionInstantTime, logCompactionMetadata, Option.of(table));
+ try {
+ HoodieTimeline pendingLogCompactionTimeline = table.getActiveTimeline().filterPendingLogCompactionTimeline();
+ if (pendingLogCompactionTimeline.containsInstant(inflightInstant)) {
+ log.info("Found Log compaction inflight file. Rolling back the commit and exiting.");
+ table.rollbackInflightLogCompaction(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), txnManager);
+ table.getMetaClient().reloadActiveTimeline();
+ throw new HoodieException("Execution is aborted since it found an Inflight logcompaction,"
+ + "log compaction plans are mutable plans, so reschedule another logcompaction.");
+ }
+ logCompactionTimer = metrics.getLogCompactionCtx();
+ WriteMarkersFactory.get(config.getMarkersType(), table, logCompactionInstantTime);
+ HoodieWriteMetadata<T> writeMetadata = table.logCompact(context, logCompactionInstantTime);
+ HoodieWriteMetadata<T> updatedWriteMetadata = partialUpdateTableMetadata(table, writeMetadata, logCompactionInstantTime, WriteOperationType.LOG_COMPACT);
+ HoodieWriteMetadata<O> logCompactionMetadata = convertToOutputMetadata(updatedWriteMetadata);
+ if (shouldComplete) {
+ commitLogCompaction(logCompactionInstantTime, logCompactionMetadata, Option.of(table));
+ }
Review Comment:
🤖 nit: The `throw e;` within the `catch` block is good. It might be helpful to log the exception at `ERROR` level here, as this indicates a failure in the log compaction process that should be explicitly captured for debugging.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -224,25 +224,46 @@ protected HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime, boo
+ "timestamp exists. Instant details %s", compactionInstantWithGreaterTimestamp.get()));
}
- HoodieTimeline pendingLogCompactionTimeline = table.getActiveTimeline().filterPendingLogCompactionTimeline();
InstantGenerator instantGenerator = table.getMetaClient().getInstantGenerator();
HoodieInstant inflightInstant = instantGenerator.getLogCompactionInflightInstant(logCompactionInstantTime);
- if (pendingLogCompactionTimeline.containsInstant(inflightInstant)) {
- log.info("Found Log compaction inflight file. Rolling back the commit and exiting.");
- table.rollbackInflightLogCompaction(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), txnManager);
- table.getMetaClient().reloadActiveTimeline();
- throw new HoodieException("Execution is aborted since it found an Inflight logcompaction,"
- + "log compaction plans are mutable plans, so reschedule another logcompaction.");
+ boolean isMultiWriter = config.getWriteConcurrencyMode().supportsMultiWriter();
+ if (isMultiWriter) {
+ try {
+ txnManager.beginStateChange(Option.of(inflightInstant), txnManager.getLastCompletedTransactionOwner());
+ validateHeartBeat(logCompactionInstantTime);
+ if (!table.getMetaClient().reloadActiveTimeline().filterPendingLogCompactionTimeline().containsInstant(logCompactionInstantTime)) {
+ throw new HoodieException("Requested log compaction instant " + logCompactionInstantTime
+ + " is not present as pending or already completed in the active timeline.");
+ }
+ this.heartbeatClient.start(logCompactionInstantTime);
+ } finally {
+ txnManager.endStateChange(Option.of(inflightInstant));
+ }
}
- logCompactionTimer = metrics.getLogCompactionCtx();
- WriteMarkersFactory.get(config.getMarkersType(), table, logCompactionInstantTime);
- HoodieWriteMetadata<T> writeMetadata = table.logCompact(context, logCompactionInstantTime);
- HoodieWriteMetadata<T> updatedWriteMetadata = partialUpdateTableMetadata(table, writeMetadata, logCompactionInstantTime, WriteOperationType.LOG_COMPACT);
- HoodieWriteMetadata<O> logCompactionMetadata = convertToOutputMetadata(updatedWriteMetadata);
- if (shouldComplete) {
- commitLogCompaction(logCompactionInstantTime, logCompactionMetadata, Option.of(table));
+ try {
+ HoodieTimeline pendingLogCompactionTimeline = table.getActiveTimeline().filterPendingLogCompactionTimeline();
+ if (pendingLogCompactionTimeline.containsInstant(inflightInstant)) {
+ log.info("Found Log compaction inflight file. Rolling back the commit and exiting.");
+ table.rollbackInflightLogCompaction(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), txnManager);
+ table.getMetaClient().reloadActiveTimeline();
+ throw new HoodieException("Execution is aborted since it found an Inflight logcompaction,"
+ + "log compaction plans are mutable plans, so reschedule another logcompaction.");
+ }
+ logCompactionTimer = metrics.getLogCompactionCtx();
+ WriteMarkersFactory.get(config.getMarkersType(), table, logCompactionInstantTime);
+ HoodieWriteMetadata<T> writeMetadata = table.logCompact(context, logCompactionInstantTime);
+ HoodieWriteMetadata<T> updatedWriteMetadata = partialUpdateTableMetadata(table, writeMetadata, logCompactionInstantTime, WriteOperationType.LOG_COMPACT);
+ HoodieWriteMetadata<O> logCompactionMetadata = convertToOutputMetadata(updatedWriteMetadata);
+ if (shouldComplete) {
Review Comment:
🤖 nit: Could we define `isMultiWriter` once at the beginning of the `logCompact` method if it's used multiple times?
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -1296,6 +1296,196 @@ public void testCompactionFailsWhenInstantNotInActiveTimeline() throws Exception
FileIOUtils.deleteDirectory(new File(basePath));
}
+ /**
+ * Test that when two writers attempt to execute the same log compaction plan concurrently,
+ * at least one will succeed and the other will fail due to heartbeat guard.
+ *
+ * This test uses a MOR table with multiwriter/optimistic concurrent control enabled.
+ */
+ @Test
+ public void testConcurrentLogCompactionExecutionOnSamePlan() throws Exception {
+ setUpMORTestTable();
+
Review Comment:
🤖 nit: The `@../hudi-review-1/hudi-utilities/src/test/java/org/apache/hudi/utilities/offlinejob/TestHoodieClusteringJob.java` annotation seems out of place here. It looks like a copy-paste error; perhaps it was intended to be a different annotation or is not needed?
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -1296,6 +1296,196 @@ public void testCompactionFailsWhenInstantNotInActiveTimeline() throws Exception
FileIOUtils.deleteDirectory(new File(basePath));
}
+ /**
+ * Test that when two writers attempt to execute the same log compaction plan concurrently,
+ * at least one will succeed and the other will fail due to heartbeat guard.
+ *
+ * This test uses a MOR table with multiwriter/optimistic concurrent control enabled.
+ */
+ @Test
+ public void testConcurrentLogCompactionExecutionOnSamePlan() throws Exception {
+ setUpMORTestTable();
+
+ Properties properties = new Properties();
+ properties.setProperty(FILESYSTEM_LOCK_PATH_PROP_KEY, basePath + "/.hoodie/.locks");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY, "3000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "1000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY, "3");
+
+ HoodieWriteConfig.Builder writeConfigBuilder = getConfigBuilder()
+ .withHeartbeatIntervalInMs(60 * 1000)
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY)
+ .withAutoClean(false).build())
Review Comment:
🤖 nit: Consider using a constant for `FILESYSTEM_LOCK_PATH_PROP_KEY` (e.g., `LockConfiguration.FILESYSTEM_LOCK_PATH_PROP_KEY.key()`) to ensure consistency and prevent typos.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -1296,6 +1296,196 @@ public void testCompactionFailsWhenInstantNotInActiveTimeline() throws Exception
FileIOUtils.deleteDirectory(new File(basePath));
}
+ /**
+ * Test that when two writers attempt to execute the same log compaction plan concurrently,
+ * at least one will succeed and the other will fail due to heartbeat guard.
+ *
+ * This test uses a MOR table with multiwriter/optimistic concurrent control enabled.
+ */
+ @Test
+ public void testConcurrentLogCompactionExecutionOnSamePlan() throws Exception {
+ setUpMORTestTable();
+
Review Comment:
🤖 nit: The Javadoc comment for this test method (`testConcurrentLogCompactionExecutionOnSamePlan`) is very good and clearly describes the test's purpose.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -458,33 +479,39 @@ public HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime) {
* Commit Log Compaction and track metrics.
*/
protected void completeLogCompaction(HoodieCommitMetadata metadata, HoodieTable table, String logCompactionCommitTime, List<HoodieWriteStat> partialMetadataWriteStats) {
Review Comment:
🤖 nit: Similar to the `logCompact` method, consider moving the `this.txnManager.beginStateChange` and `this.txnManager.endStateChange` to fully encompass the critical part of the commit, including `preCommit` and `finalizeWrite`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -224,25 +224,46 @@ protected HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime, boo
+ "timestamp exists. Instant details %s", compactionInstantWithGreaterTimestamp.get()));
}
- HoodieTimeline pendingLogCompactionTimeline = table.getActiveTimeline().filterPendingLogCompactionTimeline();
InstantGenerator instantGenerator = table.getMetaClient().getInstantGenerator();
HoodieInstant inflightInstant = instantGenerator.getLogCompactionInflightInstant(logCompactionInstantTime);
- if (pendingLogCompactionTimeline.containsInstant(inflightInstant)) {
- log.info("Found Log compaction inflight file. Rolling back the commit and exiting.");
- table.rollbackInflightLogCompaction(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), txnManager);
- table.getMetaClient().reloadActiveTimeline();
- throw new HoodieException("Execution is aborted since it found an Inflight logcompaction,"
- + "log compaction plans are mutable plans, so reschedule another logcompaction.");
+ boolean isMultiWriter = config.getWriteConcurrencyMode().supportsMultiWriter();
+ if (isMultiWriter) {
+ try {
+ txnManager.beginStateChange(Option.of(inflightInstant), txnManager.getLastCompletedTransactionOwner());
+ validateHeartBeat(logCompactionInstantTime);
+ if (!table.getMetaClient().reloadActiveTimeline().filterPendingLogCompactionTimeline().containsInstant(logCompactionInstantTime)) {
+ throw new HoodieException("Requested log compaction instant " + logCompactionInstantTime
Review Comment:
🤖 nit: Consider moving the `txnManager.beginStateChange` and `txnManager.endStateChange` into the inner `try-finally` block to keep the transaction management closely tied to the actual compaction process, potentially reducing the scope of the state change.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -1296,6 +1296,196 @@ public void testCompactionFailsWhenInstantNotInActiveTimeline() throws Exception
FileIOUtils.deleteDirectory(new File(basePath));
}
+ /**
+ * Test that when two writers attempt to execute the same log compaction plan concurrently,
+ * at least one will succeed and the other will fail due to heartbeat guard.
+ *
+ * This test uses a MOR table with multiwriter/optimistic concurrent control enabled.
+ */
+ @Test
+ public void testConcurrentLogCompactionExecutionOnSamePlan() throws Exception {
+ setUpMORTestTable();
+
+ Properties properties = new Properties();
+ properties.setProperty(FILESYSTEM_LOCK_PATH_PROP_KEY, basePath + "/.hoodie/.locks");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY, "3000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "1000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY, "3");
+
+ HoodieWriteConfig.Builder writeConfigBuilder = getConfigBuilder()
+ .withHeartbeatIntervalInMs(60 * 1000)
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY)
+ .withAutoClean(false).build())
+ .withArchivalConfig(HoodieArchivalConfig.newBuilder()
+ .withAutoArchive(false).build())
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(2)
+ .withLogCompactionBlocksThreshold(1).build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder()
+ .withStorageType(FileSystemViewStorageType.MEMORY)
+ .withSecondaryStorageType(FileSystemViewStorageType.MEMORY).build())
+ .withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
+ .withLockConfig(HoodieLockConfig.newBuilder()
+ .withLockProvider(InProcessLockProvider.class)
+ .withConflictResolutionStrategy(new SimpleConcurrentFileWritesConflictResolutionStrategy())
+ .build())
+ .withProperties(properties);
+
+ HoodieWriteConfig cfg = writeConfigBuilder.build();
+
+ SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ String firstCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithInserts(cfg, client, "000", firstCommitTime, 200);
+
+ String secondCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithUpserts(cfg, client, firstCommitTime, Option.empty(), secondCommitTime, 100);
+ String thirdCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithUpserts(cfg, client, secondCommitTime, Option.empty(), thirdCommitTime, 100);
+
+ Option<String> logCompactionInstantOpt = client.scheduleTableService(Option.empty(), Option.empty(), TableServiceType.LOG_COMPACT);
+ assertTrue(logCompactionInstantOpt.isPresent(), "Log compaction should be scheduled");
+ String logCompactionInstantTime = logCompactionInstantOpt.get();
+
+ HoodieTimeline pendingLogCompactionTimeline = metaClient.reloadActiveTimeline().filterPendingLogCompactionTimeline();
+ assertTrue(pendingLogCompactionTimeline.containsInstant(logCompactionInstantTime),
+ "Log compaction instant should be in pending state after scheduling");
+
+ final int threadCount = 2;
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+ final CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount);
+ final AtomicBoolean writer1Succeeded = new AtomicBoolean(false);
+ final AtomicBoolean writer2Succeeded = new AtomicBoolean(false);
+
+ SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
+ SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+
+ Future<?> future1 = executors.submit(() -> {
+ try {
+ cyclicBarrier.await();
+ client1.logCompact(logCompactionInstantTime, true);
+ writer1Succeeded.set(true);
+ } catch (Exception e) {
+ LOG.info("Writer 1 failed with exception: " + e.getMessage());
+ writer1Succeeded.set(false);
+ }
+ });
+
+ Future<?> future2 = executors.submit(() -> {
+ try {
+ cyclicBarrier.await();
+ client2.logCompact(logCompactionInstantTime, true);
+ writer2Succeeded.set(true);
+ } catch (Exception e) {
+ LOG.info("Writer 2 failed with exception: " + e.getMessage());
+ writer2Succeeded.set(false);
+ }
+ });
+
+ future1.get();
+ future2.get();
+
+ assertTrue(writer1Succeeded.get() || writer2Succeeded.get(),
Review Comment:
🤖 nit: Similar to the previous test, the `@../hudi-review-1/hudi-utilities/src/test/java/org/apache/hudi/utilities/offlinejob/TestHoodieClusteringJob.java` annotation appears to be a copy-paste error and should likely be removed.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -1296,6 +1296,196 @@ public void testCompactionFailsWhenInstantNotInActiveTimeline() throws Exception
FileIOUtils.deleteDirectory(new File(basePath));
}
+ /**
+ * Test that when two writers attempt to execute the same log compaction plan concurrently,
+ * at least one will succeed and the other will fail due to heartbeat guard.
+ *
+ * This test uses a MOR table with multiwriter/optimistic concurrent control enabled.
+ */
+ @Test
+ public void testConcurrentLogCompactionExecutionOnSamePlan() throws Exception {
+ setUpMORTestTable();
+
+ Properties properties = new Properties();
+ properties.setProperty(FILESYSTEM_LOCK_PATH_PROP_KEY, basePath + "/.hoodie/.locks");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY, "3000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "1000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY, "3");
+
+ HoodieWriteConfig.Builder writeConfigBuilder = getConfigBuilder()
+ .withHeartbeatIntervalInMs(60 * 1000)
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY)
+ .withAutoClean(false).build())
+ .withArchivalConfig(HoodieArchivalConfig.newBuilder()
Review Comment:
🤖 nit: For these lock configuration properties (`LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY`, `LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY`, `LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY`), using constants from `LockConfiguration` would be more robust and readable.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -1296,6 +1296,196 @@ public void testCompactionFailsWhenInstantNotInActiveTimeline() throws Exception
FileIOUtils.deleteDirectory(new File(basePath));
}
+ /**
+ * Test that when two writers attempt to execute the same log compaction plan concurrently,
+ * at least one will succeed and the other will fail due to heartbeat guard.
+ *
+ * This test uses a MOR table with multiwriter/optimistic concurrent control enabled.
+ */
+ @Test
+ public void testConcurrentLogCompactionExecutionOnSamePlan() throws Exception {
+ setUpMORTestTable();
+
+ Properties properties = new Properties();
+ properties.setProperty(FILESYSTEM_LOCK_PATH_PROP_KEY, basePath + "/.hoodie/.locks");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY, "3000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "1000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY, "3");
+
+ HoodieWriteConfig.Builder writeConfigBuilder = getConfigBuilder()
+ .withHeartbeatIntervalInMs(60 * 1000)
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY)
+ .withAutoClean(false).build())
+ .withArchivalConfig(HoodieArchivalConfig.newBuilder()
+ .withAutoArchive(false).build())
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(2)
+ .withLogCompactionBlocksThreshold(1).build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder()
+ .withStorageType(FileSystemViewStorageType.MEMORY)
+ .withSecondaryStorageType(FileSystemViewStorageType.MEMORY).build())
+ .withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
+ .withLockConfig(HoodieLockConfig.newBuilder()
+ .withLockProvider(InProcessLockProvider.class)
+ .withConflictResolutionStrategy(new SimpleConcurrentFileWritesConflictResolutionStrategy())
+ .build())
+ .withProperties(properties);
+
+ HoodieWriteConfig cfg = writeConfigBuilder.build();
+
+ SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ String firstCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithInserts(cfg, client, "000", firstCommitTime, 200);
+
+ String secondCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithUpserts(cfg, client, firstCommitTime, Option.empty(), secondCommitTime, 100);
+ String thirdCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithUpserts(cfg, client, secondCommitTime, Option.empty(), thirdCommitTime, 100);
+
+ Option<String> logCompactionInstantOpt = client.scheduleTableService(Option.empty(), Option.empty(), TableServiceType.LOG_COMPACT);
+ assertTrue(logCompactionInstantOpt.isPresent(), "Log compaction should be scheduled");
+ String logCompactionInstantTime = logCompactionInstantOpt.get();
+
+ HoodieTimeline pendingLogCompactionTimeline = metaClient.reloadActiveTimeline().filterPendingLogCompactionTimeline();
+ assertTrue(pendingLogCompactionTimeline.containsInstant(logCompactionInstantTime),
+ "Log compaction instant should be in pending state after scheduling");
+
+ final int threadCount = 2;
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+ final CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount);
+ final AtomicBoolean writer1Succeeded = new AtomicBoolean(false);
+ final AtomicBoolean writer2Succeeded = new AtomicBoolean(false);
+
+ SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
+ SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+
+ Future<?> future1 = executors.submit(() -> {
+ try {
+ cyclicBarrier.await();
+ client1.logCompact(logCompactionInstantTime, true);
+ writer1Succeeded.set(true);
+ } catch (Exception e) {
+ LOG.info("Writer 1 failed with exception: " + e.getMessage());
+ writer1Succeeded.set(false);
+ }
+ });
+
+ Future<?> future2 = executors.submit(() -> {
+ try {
+ cyclicBarrier.await();
+ client2.logCompact(logCompactionInstantTime, true);
+ writer2Succeeded.set(true);
+ } catch (Exception e) {
+ LOG.info("Writer 2 failed with exception: " + e.getMessage());
+ writer2Succeeded.set(false);
+ }
+ });
+
+ future1.get();
+ future2.get();
+
+ assertTrue(writer1Succeeded.get() || writer2Succeeded.get(),
Review Comment:
🤖 nit: The Javadoc for `testLogCompactionFailsWhenInstantNotInActiveTimeline` is well-written and clear.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -1296,6 +1296,196 @@ public void testCompactionFailsWhenInstantNotInActiveTimeline() throws Exception
FileIOUtils.deleteDirectory(new File(basePath));
}
+ /**
+ * Test that when two writers attempt to execute the same log compaction plan concurrently,
+ * at least one will succeed and the other will fail due to heartbeat guard.
+ *
+ * This test uses a MOR table with multiwriter/optimistic concurrent control enabled.
+ */
+ @Test
+ public void testConcurrentLogCompactionExecutionOnSamePlan() throws Exception {
+ setUpMORTestTable();
+
+ Properties properties = new Properties();
+ properties.setProperty(FILESYSTEM_LOCK_PATH_PROP_KEY, basePath + "/.hoodie/.locks");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY, "3000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "1000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY, "3");
+
+ HoodieWriteConfig.Builder writeConfigBuilder = getConfigBuilder()
+ .withHeartbeatIntervalInMs(60 * 1000)
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY)
+ .withAutoClean(false).build())
+ .withArchivalConfig(HoodieArchivalConfig.newBuilder()
+ .withAutoArchive(false).build())
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(2)
+ .withLogCompactionBlocksThreshold(1).build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder()
+ .withStorageType(FileSystemViewStorageType.MEMORY)
+ .withSecondaryStorageType(FileSystemViewStorageType.MEMORY).build())
+ .withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
+ .withLockConfig(HoodieLockConfig.newBuilder()
+ .withLockProvider(InProcessLockProvider.class)
+ .withConflictResolutionStrategy(new SimpleConcurrentFileWritesConflictResolutionStrategy())
+ .build())
+ .withProperties(properties);
+
+ HoodieWriteConfig cfg = writeConfigBuilder.build();
+
+ SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ String firstCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithInserts(cfg, client, "000", firstCommitTime, 200);
+
+ String secondCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithUpserts(cfg, client, firstCommitTime, Option.empty(), secondCommitTime, 100);
+ String thirdCommitTime = WriteClientTestUtils.createNewInstantTime();
+ createCommitWithUpserts(cfg, client, secondCommitTime, Option.empty(), thirdCommitTime, 100);
+
+ Option<String> logCompactionInstantOpt = client.scheduleTableService(Option.empty(), Option.empty(), TableServiceType.LOG_COMPACT);
+ assertTrue(logCompactionInstantOpt.isPresent(), "Log compaction should be scheduled");
+ String logCompactionInstantTime = logCompactionInstantOpt.get();
+
+ HoodieTimeline pendingLogCompactionTimeline = metaClient.reloadActiveTimeline().filterPendingLogCompactionTimeline();
+ assertTrue(pendingLogCompactionTimeline.containsInstant(logCompactionInstantTime),
+ "Log compaction instant should be in pending state after scheduling");
+
+ final int threadCount = 2;
+ final ExecutorService executors = Executors.newFixedThreadPool(threadCount);
+ final CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount);
+ final AtomicBoolean writer1Succeeded = new AtomicBoolean(false);
+ final AtomicBoolean writer2Succeeded = new AtomicBoolean(false);
+
+ SparkRDDWriteClient client1 = getHoodieWriteClient(cfg);
+ SparkRDDWriteClient client2 = getHoodieWriteClient(cfg);
+
+ Future<?> future1 = executors.submit(() -> {
+ try {
+ cyclicBarrier.await();
+ client1.logCompact(logCompactionInstantTime, true);
+ writer1Succeeded.set(true);
+ } catch (Exception e) {
+ LOG.info("Writer 1 failed with exception: " + e.getMessage());
+ writer1Succeeded.set(false);
+ }
+ });
+
+ Future<?> future2 = executors.submit(() -> {
+ try {
+ cyclicBarrier.await();
+ client2.logCompact(logCompactionInstantTime, true);
+ writer2Succeeded.set(true);
+ } catch (Exception e) {
+ LOG.info("Writer 2 failed with exception: " + e.getMessage());
+ writer2Succeeded.set(false);
+ }
+ });
+
+ future1.get();
+ future2.get();
+
+ assertTrue(writer1Succeeded.get() || writer2Succeeded.get(),
+ "At least one writer should succeed in executing the log compaction");
+
+ assertTrue(writer1Succeeded.get() ^ writer2Succeeded.get(),
+ "Exactly one writer should succeed in executing the log compaction");
+
+ executors.shutdown();
+ client.close();
+ client1.close();
+ client2.close();
+ FileIOUtils.deleteDirectory(new File(basePath));
+ }
Review Comment:
🤖 nit: Consider using constants from `LockConfiguration` for these properties as well.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/TestHoodieClientMultiWriter.java:
##########
@@ -1296,6 +1296,196 @@ public void testCompactionFailsWhenInstantNotInActiveTimeline() throws Exception
FileIOUtils.deleteDirectory(new File(basePath));
}
+ /**
+ * Test that when two writers attempt to execute the same log compaction plan concurrently,
+ * at least one will succeed and the other will fail due to heartbeat guard.
+ *
+ * This test uses a MOR table with multiwriter/optimistic concurrent control enabled.
+ */
+ @Test
+ public void testConcurrentLogCompactionExecutionOnSamePlan() throws Exception {
+ setUpMORTestTable();
+
+ Properties properties = new Properties();
+ properties.setProperty(FILESYSTEM_LOCK_PATH_PROP_KEY, basePath + "/.hoodie/.locks");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY, "3000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "1000");
+ properties.setProperty(LockConfiguration.LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY, "3");
+
+ HoodieWriteConfig.Builder writeConfigBuilder = getConfigBuilder()
+ .withHeartbeatIntervalInMs(60 * 1000)
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY)
+ .withAutoClean(false).build())
+ .withArchivalConfig(HoodieArchivalConfig.newBuilder()
+ .withAutoArchive(false).build())
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(2)
+ .withLogCompactionBlocksThreshold(1).build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder()
+ .withStorageType(FileSystemViewStorageType.MEMORY)
+ .withSecondaryStorageType(FileSystemViewStorageType.MEMORY).build())
+ .withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
+ .withLockConfig(HoodieLockConfig.newBuilder()
+ .withLockProvider(InProcessLockProvider.class)
+ .withConflictResolutionStrategy(new SimpleConcurrentFileWritesConflictResolutionStrategy())
+ .build())
+ .withProperties(properties);
+
+ HoodieWriteConfig cfg = writeConfigBuilder.build();
+
+ SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ String firstCommitTime = WriteClientTestUtils.createNewInstantTime();
Review Comment:
🤖 nit: It's good practice to close the `ExecutorService` explicitly using `executors.shutdownNow()` (or `shutdown()` then `awaitTermination()`) to ensure all threads are terminated gracefully and resources are released, especially in a test context.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org