This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e70f8301601699173cf74290610bbdfd67077277 Author: Sivabalan Narayanan <[email protected]> AuthorDate: Fri Jun 26 16:49:29 2026 -0700 [HUDI-18827] Fix per-task write token for MOR (table v6) rollback log files (#18828) fix(rollback): use per-task write tokens for MOR (table v6) rollback log files RollbackHelperV1 was overriding the per-task write token with the existing log file's token (often UNKNOWN_WRITE_TOKEN), causing retried rollbacks to collide on file name. Keep the per-task token and bump log version to latest + 1 instead. --------- Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]> (cherry picked from commit 17a76973a34e2430e7adc878888f07abd1d0b6d2) --- .../table/action/rollback/RollbackHelperV1.java | 35 ++- .../table/action/rollback/TestRollbackHelper.java | 25 ++- .../TestMarkerBasedRollbackStrategy.java | 16 +- .../TestMergeOnReadRollbackActionExecutor.java | 235 +++++++++++++++++++++ 4 files changed, 292 insertions(+), 19 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java index 1b288a70e8ba..b24d85dd9e8d 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java @@ -141,7 +141,12 @@ public class RollbackHelperV1 extends RollbackHelper { } } - Pair<Integer, String> sentinel = Pair.of(HoodieLogFile.LOGFILE_BASE_VERSION, HoodieLogFormat.UNKNOWN_WRITE_TOKEN); + // Insert a sentinel for file groups with no existing log files so the caller can skip a + // redundant per-request FS listing. The sentinel uses a null write-token to distinguish + // it from a real entry whose token happens to equal UNKNOWN_WRITE_TOKEN; otherwise tests + // (and any callers that previously wrote logs with the legacy "1-0-1" token) would be + // indistinguishable from "no log file present". + Pair<Integer, String> sentinel = Pair.of(HoodieLogFile.LOGFILE_BASE_VERSION, null); for (String expectedKey : expectedKeys) { logVersionMap.putIfAbsent(expectedKey, sentinel); } @@ -324,16 +329,30 @@ public class RollbackHelperV1 extends RollbackHelper { .withTableVersion(tableVersion) .withFileExtension(HoodieLogFile.DELTA_EXTENSION); - // Supply the pre-computed latest log version and its write token so that - // WriterBuilder.build() skips the per-request FSUtils.getLatestLogVersion() listing. - // This produces the same result: build() would discover (N, T_existing), construct - // path (N, T_existing), find it exists, and roll over to N+1. Pre-computation - // feeds the same (N, T_existing), triggering the identical rollover in getOutputStream(). + // Apply pre-computed log version if available. Always keep the per-task write token + // generated above (via CommonClientUtils.generateWriteToken) so that retried/repeated + // rollbacks do not collide on UNKNOWN_WRITE_TOKEN or inherit a prior log's write token. + // + // When doDelete=true, we actually create a new rollback log file: explicitly bump the + // version (latest + 1) so the new file is written with the per-task write token instead + // of rolling over and inheriting the existing log file's token. When doDelete=false we + // are only collecting stats (no append), so we let WriterBuilder.build() discover the + // existing version itself — bumping here would point to a non-existent path and break + // the downstream storage.getPathInfo lookup. + // + // The sentinel value (right == null) means "partition listed but no log file for this + // file group" — write at the base version. String logVersionKey = logVersionLookupKey(partitionPath, fileId, rollbackRequest.getLatestBaseInstant()); Pair<Integer, String> preComputedVersion = logVersionMap.get(logVersionKey); if (preComputedVersion != null) { - writerBuilder.withLogVersion(preComputedVersion.getLeft()) - .withLogWriteToken(preComputedVersion.getRight()); + if (preComputedVersion.getRight() == null) { + writerBuilder.withLogVersion(HoodieLogFile.LOGFILE_BASE_VERSION); + } else if (doDelete) { + writerBuilder.withLogVersion(preComputedVersion.getLeft() + 1); + } else { + writerBuilder.withLogVersion(preComputedVersion.getLeft()) + .withLogWriteToken(preComputedVersion.getRight()); + } } writer = writerBuilder.build(); diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackHelper.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackHelper.java index 15c2421cd6df..705aec57e66d 100644 --- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackHelper.java +++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackHelper.java @@ -31,7 +31,6 @@ import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.table.log.HoodieLogFormat; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; -import org.apache.hudi.common.testutils.FileCreateUtils; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.common.util.collection.Triple; @@ -64,6 +63,7 @@ import java.util.stream.IntStream; import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -681,7 +681,9 @@ class TestRollbackHelper extends HoodieRollbackTestBase { String missingKey = RollbackHelperV1.logVersionLookupKey(partition, "fileId-no-logs", baseInstant); assertTrue(result.containsKey(missingKey)); assertEquals(HoodieLogFile.LOGFILE_BASE_VERSION, (int) result.get(missingKey).getLeft()); - assertEquals(HoodieLogFormat.UNKNOWN_WRITE_TOKEN, result.get(missingKey).getRight()); + // Sentinel entries (no real log file) carry a null write token so they cannot be confused + // with a real log file that happens to use UNKNOWN_WRITE_TOKEN. + assertNull(result.get(missingKey).getRight()); } @Test @@ -698,10 +700,15 @@ class TestRollbackHelper extends HoodieRollbackTestBase { ctx.rollbackRequests, true, 5); validateStateAfterRollback(ctx.rollbackRequests); + // Rollback log files are written with a per-task write token from TaskContextSupplier. + // HoodieLocalEngineContext uses LocalTaskContextSupplier which returns 0/0/0 -> token "0-0-0". + String rollbackWriteToken = FSUtils.makeWriteToken(0, 0, 0); StoragePath rollbackLogPath1 = new StoragePath(new StoragePath(basePath, ctx.partition2), - FileCreateUtils.logFileName(ctx.baseInstantTimeOfLogFiles, ctx.logFileId1, 2)); + FSUtils.makeLogFileName(ctx.logFileId1, HoodieLogFile.DELTA_EXTENSION, + ctx.baseInstantTimeOfLogFiles, 2, rollbackWriteToken)); StoragePath rollbackLogPath2 = new StoragePath(new StoragePath(basePath, ctx.partition2), - FileCreateUtils.logFileName(ctx.baseInstantTimeOfLogFiles, ctx.logFileId2, ROLLBACK_LOG_VERSION)); + FSUtils.makeLogFileName(ctx.logFileId2, HoodieLogFile.DELTA_EXTENSION, + ctx.baseInstantTimeOfLogFiles, ROLLBACK_LOG_VERSION, rollbackWriteToken)); List<Pair<String, HoodieRollbackStat>> expected = buildExpectedBaseFileStats(ctx); expected.add(Pair.of(ctx.partition2, @@ -733,8 +740,10 @@ class TestRollbackHelper extends HoodieRollbackTestBase { ctx.rollbackRequests, true, 5); validateStateAfterRollback(ctx.rollbackRequests); + String rollbackWriteToken = FSUtils.makeWriteToken(0, 0, 0); StoragePath rollbackLogPath = new StoragePath(new StoragePath(basePath, ctx.partition), - FileCreateUtils.logFileName(ctx.baseInstantTimeOfLogFiles, ctx.logFileId, ROLLBACK_LOG_VERSION)); + FSUtils.makeLogFileName(ctx.logFileId, HoodieLogFile.DELTA_EXTENSION, + ctx.baseInstantTimeOfLogFiles, ROLLBACK_LOG_VERSION, rollbackWriteToken)); List<Pair<String, HoodieRollbackStat>> expected = new ArrayList<>(); expected.add(Pair.of(ctx.partition, @@ -797,8 +806,12 @@ class TestRollbackHelper extends HoodieRollbackTestBase { assertTrue(storage.exists(new StoragePath(partitionStoragePath, logFileName))); } + // doDelete=false: no rollback log file is created. The reported path is the existing latest + // log file (the WriterBuilder rediscovers the existing log when we don't explicitly bump the + // version), so it carries the existing log file's write token. StoragePath rollbackLogPath = new StoragePath(partitionStoragePath, - FileCreateUtils.logFileName(ctx.baseInstantTimeOfLogFiles, ctx.logFileId, ctx.logVersionCount)); + FSUtils.makeLogFileName(ctx.logFileId, HoodieLogFile.DELTA_EXTENSION, + ctx.baseInstantTimeOfLogFiles, ctx.logVersionCount, HoodieLogFormat.UNKNOWN_WRITE_TOKEN)); List<Pair<String, HoodieRollbackStat>> expected = Collections.singletonList( Pair.of(ctx.partition, HoodieRollbackStat.newBuilder() diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestMarkerBasedRollbackStrategy.java index 182f65100b20..bd9ddb23bf03 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestMarkerBasedRollbackStrategy.java @@ -30,6 +30,7 @@ import org.apache.hudi.client.embedded.EmbeddedTimelineService; import org.apache.hudi.common.HoodieRollbackStat; import org.apache.hudi.common.config.HoodieMetadataConfig; import org.apache.hudi.common.model.HoodieFileFormat; +import org.apache.hudi.common.model.HoodieLogFile; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.IOType; @@ -43,6 +44,7 @@ import org.apache.hudi.common.testutils.HoodieTestTable; import org.apache.hudi.common.testutils.HoodieTestUtils; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.storage.StoragePathInfo; import org.apache.hudi.table.HoodieSparkTable; import org.apache.hudi.table.HoodieTable; import org.apache.hudi.table.action.rollback.MarkerBasedRollbackStrategy; @@ -399,12 +401,16 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase { assertEquals(1, rollbackStats.size()); HoodieRollbackStat rollbackStat = rollbackStats.get(0); if (!tableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT)) { - StoragePath rollbackLogPath = new StoragePath(new StoragePath(basePath, partition), - FileCreateUtils.logFileName(instantTime1, fileId, numLogFiles + 2)); + // The rollback log file's write token is determined by Spark's task context at runtime, + // so extract the actual path from the rollback stats rather than constructing it with a + // hardcoded write token. Verify the file exists and its version matches the expected bump. + StoragePathInfo rollbackLogPathInfo = + rollbackStat.getCommandBlocksCount().entrySet().stream().findFirst().get().getKey(); + StoragePath rollbackLogPath = rollbackLogPathInfo.getPath(); assertTrue(storage.exists(rollbackLogPath)); - assertEquals(rollbackLogPath.getPathWithoutSchemeAndAuthority(), - rollbackStat.getCommandBlocksCount().entrySet().stream().findFirst().get() - .getKey().getPath().getPathWithoutSchemeAndAuthority()); + HoodieLogFile rollbackLogFile = new HoodieLogFile(rollbackLogPathInfo); + assertEquals(fileId, rollbackLogFile.getFileId()); + assertEquals(numLogFiles + 2, rollbackLogFile.getLogVersion()); } assertEquals(partition, rollbackStat.getPartitionPath()); assertEquals( diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java index 7248078bbb88..467531e89a93 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java +++ b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java @@ -21,10 +21,12 @@ package org.apache.hudi.table.action.rollback; import org.apache.hudi.avro.model.HoodieRollbackMetadata; import org.apache.hudi.avro.model.HoodieRollbackPartitionMetadata; import org.apache.hudi.client.SparkRDDWriteClient; +import org.apache.hudi.client.WriteClientTestUtils; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.config.HoodieMetadataConfig; import org.apache.hudi.common.config.HoodieStorageConfig; import org.apache.hudi.common.fs.ConsistencyGuardConfig; +import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.model.FileSlice; import org.apache.hudi.common.model.HoodieCommitMetadata; import org.apache.hudi.common.model.HoodieFileGroup; @@ -32,6 +34,10 @@ import org.apache.hudi.common.model.HoodieLogFile; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.common.table.marker.MarkerType; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion; @@ -52,6 +58,12 @@ import org.apache.hudi.table.marker.WriteMarkersFactory; import org.apache.hudi.testutils.Assertions; import org.apache.hudi.testutils.MetadataMergeWriteStatus; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.FileUtil; +import org.apache.hadoop.fs.LocatedFileStatus; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RemoteIterator; import org.apache.spark.api.java.JavaRDD; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -62,17 +74,21 @@ import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Properties; import java.util.stream.Collectors; import java.util.stream.Stream; import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH; import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH; import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH; +import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_FILE_NAME_GENERATOR; import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; public class TestMergeOnReadRollbackActionExecutor extends HoodieClientRollbackTestBase { @@ -487,4 +503,223 @@ public class TestMergeOnReadRollbackActionExecutor extends HoodieClientRollbackT client.rollback(newCommitTime); } } + + /** + * Tests that rollback operations generate unique write tokens for log files, preventing collisions + * during repeated rollback attempts. + * + * <p>This test validates the fix for write token generation in metadata table rollbacks. Previously, + * rollback log files used the default UNKNOWN_WRITE_TOKEN ("1-0-1"), causing collisions when rollback + * was retried. Now, each rollback generates explicit write tokens based on Spark task context + * (format: {partitionId}-{stageId}-{attemptId}). + * + * <p>Test flow: + * <ol> + * <li>Create initial commit with inserts to establish base files</li> + * <li>Create second commit with updates to generate log files (MOR table)</li> + * <li>Backup commit timeline files and marker directory for repeated rollback simulation</li> + * <li>Execute first rollback and validate write tokens are NOT "1-0-1"</li> + * <li>Restore commit state (timeline files + markers) to simulate rollback retry scenario</li> + * <li>Execute second rollback and validate unique write tokens prevent collisions</li> + * <li>Verify exactly one new rollback log file per file group from second attempt</li> + * </ol> + * + * @param enableMetadataTable runs the test both with and without metadata table enabled to + * ensure write-token generation is correct in both code paths + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testRollbackWriteTokenGeneration(boolean enableMetadataTable) throws Exception { + // 1. Setup: Create a table-version-6 MOR table so the rollback exercises RollbackHelperV1 + // (which is what this test targets). On v8+ rollbacks delete files directly and don't + // produce rollback log files. + Properties props = new Properties(); + props.put(HoodieTableConfig.VERSION.key(), HoodieTableVersion.SIX.versionCode()); + tearDown(); + initPath(); + initSparkContexts(); + dataGen = new HoodieTestDataGenerator( + new String[] {DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH}); + initHoodieStorage(); + initMetaClient(HoodieTableType.MERGE_ON_READ, props); + + HoodieWriteConfig cfg = getConfigBuilder() + .withRollbackUsingMarkers(true) + .withMarkersType(MarkerType.DIRECT.name()) + .withWriteTableVersion(HoodieTableVersion.SIX.versionCode()) + .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(enableMetadataTable).build()) + .withCompactionConfig(HoodieCompactionConfig.newBuilder().compactionSmallFileSize(0).build()) + .build(); + + HoodieTestDataGenerator.writePartitionMetadataDeprecated( + storage, new String[] {DEFAULT_FIRST_PARTITION_PATH}, basePath); + FileSystem fs = (FileSystem) storage.getFileSystem(); + SparkRDDWriteClient client = getHoodieWriteClient(cfg); + + // Write 1: Initial inserts + String commitTime1 = "001"; + WriteClientTestUtils.startCommitWithTime(client, commitTime1); + List<HoodieRecord> records = dataGen.generateInsertsForPartition(commitTime1, 100, DEFAULT_FIRST_PARTITION_PATH); + JavaRDD<HoodieRecord> writeRecords = jsc.parallelize(records, 1); + List<WriteStatus> statusList = client.upsert(writeRecords, commitTime1).collect(); + Assertions.assertNoWriteErrors(statusList); + client.commit(commitTime1, jsc.parallelize(statusList)); + + // Write 2: Updates to same partition to create log files. Use multiple Spark partitions to + // exercise multiple task contexts (so write tokens vary across tasks). + String commitTime2 = "002"; + WriteClientTestUtils.startCommitWithTime(client, commitTime2); + List<HoodieRecord> updateRecords = dataGen.generateUpdates(commitTime2, records); + writeRecords = jsc.parallelize(updateRecords, 2); + statusList = client.upsert(writeRecords, commitTime2).collect(); + Assertions.assertNoWriteErrors(statusList); + // Intentionally leave commit 002 in inflight state so rollback exercises the inflight path. + + HoodieTable table = this.getHoodieTable(metaClient, cfg); + Map<String, List<String>> logFileNames = collectLogFileNamesByFileId(fs, DEFAULT_FIRST_PARTITION_PATH); + assertFalse(logFileNames.isEmpty()); + + // Backup commit 002 timeline files + marker dir so the rollback retry below can replay the same input. + Path commit2RequestedPath = new Path(metaClient.getMetaPath().toString(), + commitTime2 + HoodieTimeline.REQUESTED_DELTA_COMMIT_EXTENSION); + Path commit2InflightPath = new Path(metaClient.getMetaPath().toString(), + commitTime2 + HoodieTimeline.INFLIGHT_DELTA_COMMIT_EXTENSION); + Path commit2MarkerDir = new Path(metaClient.getMarkerFolderPath(commitTime2)); + Path backupDir = new Path(basePath, ".backup_test"); + Path backupMarkerDir = new Path(backupDir, commitTime2); + fs.mkdirs(backupDir); + + boolean requestedExists = fs.exists(commit2RequestedPath); + boolean inflightExists = fs.exists(commit2InflightPath); + boolean markerDirExists = fs.exists(commit2MarkerDir); + + if (requestedExists) { + FileUtil.copy(fs, commit2RequestedPath, fs, + new Path(backupDir, commitTime2 + HoodieTimeline.REQUESTED_DELTA_COMMIT_EXTENSION), + false, fs.getConf()); + } + if (inflightExists) { + FileUtil.copy(fs, commit2InflightPath, fs, + new Path(backupDir, commitTime2 + HoodieTimeline.INFLIGHT_DELTA_COMMIT_EXTENSION), + false, fs.getConf()); + } + if (markerDirExists) { + FileUtil.copy(fs, commit2MarkerDir, fs, backupMarkerDir, false, fs.getConf()); + } + + // 3. Rollback commit 002 + String rollbackTime = "003"; + HoodieInstant rollBackInstant = INSTANT_GENERATOR.createNewInstant( + HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, commitTime2); + BaseRollbackPlanActionExecutor rollbackPlanExecutor = new BaseRollbackPlanActionExecutor( + context, cfg, table, rollbackTime, rollBackInstant, false, true, false); + rollbackPlanExecutor.execute().get(); + + MergeOnReadRollbackActionExecutor rollbackExecutor = new MergeOnReadRollbackActionExecutor( + context, cfg, table, rollbackTime, rollBackInstant, true, false); + Map<String, HoodieRollbackPartitionMetadata> rollbackMetadata = rollbackExecutor.execute().getPartitionMetadata(); + + assertEquals(1, rollbackMetadata.size()); + HoodieRollbackPartitionMetadata partitionMetadata = rollbackMetadata.get(DEFAULT_FIRST_PARTITION_PATH); + assertFalse(partitionMetadata.getRollbackLogFiles().isEmpty(), "Should have rollback log files"); + + metaClient = HoodieTableMetaClient.reload(metaClient); + table = this.getHoodieTable(metaClient, cfg); + + // Validate write tokens on the rollback log files are per-task generated (not UNKNOWN_WRITE_TOKEN "1-0-1"). + List<FileSlice> rollbackFileSlices = table.getSliceView() + .getLatestFileSlices(DEFAULT_FIRST_PARTITION_PATH) + .collect(Collectors.toList()); + // FileSlice.getLogFiles() is sorted highest-version first (reverse comparator), + // so index 0 is the latest log file produced by the rollback. + List<HoodieLogFile> rollbackLogFiles = rollbackFileSlices.stream() + .flatMap(slice -> { + List<HoodieLogFile> logFiles = slice.getLogFiles().collect(Collectors.toList()); + return Collections.singleton(logFiles.get(0)).stream(); + }) + .collect(Collectors.toList()); + + assertFalse(rollbackLogFiles.isEmpty(), "Should have rollback log files with rollback instant time"); + for (HoodieLogFile logFile : rollbackLogFiles) { + String writeToken = logFile.getLogWriteToken(); + assertFalse(writeToken.isEmpty(), "Write token should not be empty"); + assertTrue(writeToken.matches("\\d+-\\d+-\\d+"), + String.format("Write token should match pattern partitionId-stageId-attemptId, but got: %s in file: %s", + writeToken, logFile.getFileName())); + assertNotEquals("1-0-1", writeToken); + } + + Map<String, List<String>> logFileNamesPostRollback = collectLogFileNamesByFileId(fs, DEFAULT_FIRST_PARTITION_PATH); + + // Simulate rollback retry: remove rollback timeline files and restore commit 002 timeline + markers. + HoodieInstant lastRollbackInstant = metaClient.getActiveTimeline().getRollbackTimeline().lastInstant().get(); + String latestRollbackCompletedFileName = + INSTANT_FILE_NAME_GENERATOR.getFileName(lastRollbackInstant); + fs.delete(new Path(metaClient.getMetaPath().toString(), latestRollbackCompletedFileName), false); + fs.delete(new Path(metaClient.getMetaPath().toString(), + rollbackTime + HoodieTimeline.INFLIGHT_ROLLBACK_EXTENSION), false); + + if (requestedExists) { + FileUtil.copy(fs, new Path(backupDir, commitTime2 + HoodieTimeline.REQUESTED_DELTA_COMMIT_EXTENSION), + fs, commit2RequestedPath, false, fs.getConf()); + } + if (inflightExists) { + FileUtil.copy(fs, new Path(backupDir, commitTime2 + HoodieTimeline.INFLIGHT_DELTA_COMMIT_EXTENSION), + fs, commit2InflightPath, false, fs.getConf()); + } + if (markerDirExists) { + FileUtil.copy(fs, backupMarkerDir, fs, commit2MarkerDir.getParent(), false, fs.getConf()); + } + fs.delete(backupDir, true); + + metaClient = HoodieTableMetaClient.reload(metaClient); + table = this.getHoodieTable(metaClient, cfg); + + // Trigger second rollback - should create additional rollback log files with different write tokens. + MergeOnReadRollbackActionExecutor rollbackExecutor2 = new MergeOnReadRollbackActionExecutor( + context, cfg, table, rollbackTime, rollBackInstant, true, false); + Map<String, HoodieRollbackPartitionMetadata> rollbackMetadata2 = rollbackExecutor2.execute().getPartitionMetadata(); + + assertEquals(1, rollbackMetadata2.size()); + HoodieRollbackPartitionMetadata partitionMetadata2 = rollbackMetadata2.get(DEFAULT_FIRST_PARTITION_PATH); + assertFalse(partitionMetadata2.getRollbackLogFiles().isEmpty(), "Should have rollback log files"); + + metaClient = HoodieTableMetaClient.reload(metaClient); + + Map<String, List<String>> logFileNamesPost2ndRollback = collectLogFileNamesByFileId(fs, DEFAULT_FIRST_PARTITION_PATH); + Map<String, Integer> filesFrom2ndRollback = new HashMap<>(); + logFileNamesPost2ndRollback.forEach((fileId, fileNames) -> { + List<String> previousFiles = logFileNamesPostRollback.getOrDefault(fileId, Collections.emptyList()); + for (String fileName : fileNames) { + if (!previousFiles.contains(fileName)) { + filesFrom2ndRollback.merge(fileId, 1, Integer::sum); + assertNotEquals("1-0-1", new HoodieLogFile(fileName).getLogWriteToken()); + } + } + }); + + assertFalse(filesFrom2ndRollback.isEmpty(), + "Second rollback should produce at least one new log file (no collision with first rollback)"); + assertEquals(logFileNames.size(), filesFrom2ndRollback.size()); + filesFrom2ndRollback.forEach((k, v) -> assertEquals(1, v)); + client.close(); + } + + /** + * Lists all log files in the given partition and groups their file names by file ID. + */ + private Map<String, List<String>> collectLogFileNamesByFileId(FileSystem fs, String partitionPath) throws IOException { + Map<String, List<String>> logFilesByFileId = new HashMap<>(); + RemoteIterator<LocatedFileStatus> itr = fs.listFiles( + new Path(metaClient.getBasePath().toString() + "/" + partitionPath), false); + while (itr.hasNext()) { + FileStatus fileStatus = itr.next(); + String fileName = fileStatus.getPath().getName(); + if (FSUtils.isLogFile(fileName)) { + String fileId = FSUtils.getFileId(fileName); + logFilesByFileId.computeIfAbsent(fileId, k -> new ArrayList<>()).add(fileName); + } + } + return logFilesByFileId; + } }
