This is an automated email from the ASF dual-hosted git repository.
nsivabalan 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 17a76973a34e [HUDI-18827] Fix per-task write token for MOR (table v6)
rollback log files (#18828)
17a76973a34e is described below
commit 17a76973a34e2430e7adc878888f07abd1d0b6d2
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]>
---
.../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;
+ }
}
