Skip to content

Commit 8479da5

Browse files
codopecodope
committed
[HUDI-4550] Fallback to listing based rollback for completed instant
1 parent fcdd4cf commit 8479da5

9 files changed

Lines changed: 64 additions & 47 deletions

File tree

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/CopyOnWriteRestoreActionExecutor.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,6 @@ protected HoodieRollbackMetadata rollbackInstant(HoodieInstant instantToRollback
5858
instantToRollback,
5959
true,
6060
true,
61-
false,
6261
false);
6362
return rollbackActionExecutor.execute();
6463
}

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/MergeOnReadRestoreActionExecutor.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,6 @@ protected HoodieRollbackMetadata rollbackInstant(HoodieInstant instantToRollback
6262
instantToRollback,
6363
true,
6464
true,
65-
false,
6665
false);
6766

6867
// TODO : Get file status and create a rollback stat and file

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java‎

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,6 @@ public abstract class BaseRollbackActionExecutor<T extends HoodieRecordPayload,
5757
protected final HoodieInstant instantToRollback;
5858
protected final boolean deleteInstants;
5959
protected final boolean skipTimelinePublish;
60-
protected final boolean useMarkerBasedStrategy;
6160
private final TransactionManager txnManager;
6261
private final boolean skipLocking;
6362

@@ -70,8 +69,7 @@ public BaseRollbackActionExecutor(HoodieEngineContext context,
7069
HoodieInstant instantToRollback,
7170
boolean deleteInstants,
7271
boolean skipLocking) {
73-
this(context, config, table, instantTime, instantToRollback, deleteInstants,
74-
false, config.shouldRollbackUsingMarkers(), skipLocking);
72+
this(context, config, table, instantTime, instantToRollback, deleteInstants, false, skipLocking);
7573
}
7674

7775
public BaseRollbackActionExecutor(HoodieEngineContext context,
@@ -81,18 +79,12 @@ public BaseRollbackActionExecutor(HoodieEngineContext context,
8179
HoodieInstant instantToRollback,
8280
boolean deleteInstants,
8381
boolean skipTimelinePublish,
84-
boolean useMarkerBasedStrategy,
8582
boolean skipLocking) {
8683
super(context, config, table, instantTime);
8784
this.instantToRollback = instantToRollback;
8885
this.resolvedInstant = instantToRollback;
8986
this.deleteInstants = deleteInstants;
9087
this.skipTimelinePublish = skipTimelinePublish;
91-
this.useMarkerBasedStrategy = useMarkerBasedStrategy;
92-
if (useMarkerBasedStrategy) {
93-
ValidationUtils.checkArgument(!instantToRollback.isCompleted(),
94-
"Cannot use marker based rollback strategy on completed instant:" + instantToRollback);
95-
}
9688
this.skipLocking = skipLocking;
9789
this.txnManager = new TransactionManager(config, table.getMetaClient().getFs());
9890
}

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackPlanActionExecutor.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ public BaseRollbackPlanActionExecutor(HoodieEngineContext context,
6565
super(context, config, table, instantTime);
6666
this.instantToRollback = instantToRollback;
6767
this.skipTimelinePublish = skipTimelinePublish;
68-
this.shouldRollbackUsingMarkers = shouldRollbackUsingMarkers;
68+
this.shouldRollbackUsingMarkers = shouldRollbackUsingMarkers && !instantToRollback.isCompleted();
6969
}
7070

7171
/**

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/CopyOnWriteRollbackActionExecutor.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,8 @@ public CopyOnWriteRollbackActionExecutor(HoodieEngineContext context,
5555
HoodieInstant commitInstant,
5656
boolean deleteInstants,
5757
boolean skipTimelinePublish,
58-
boolean useMarkerBasedStrategy,
5958
boolean skipLocking) {
60-
super(context, config, table, instantTime, commitInstant, deleteInstants, skipTimelinePublish, useMarkerBasedStrategy, skipLocking);
59+
super(context, config, table, instantTime, commitInstant, deleteInstants, skipTimelinePublish, skipLocking);
6160
}
6261

6362
@Override

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/MergeOnReadRollbackActionExecutor.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,8 @@ public MergeOnReadRollbackActionExecutor(HoodieEngineContext context,
5555
HoodieInstant commitInstant,
5656
boolean deleteInstants,
5757
boolean skipTimelinePublish,
58-
boolean useMarkerBasedStrategy,
5958
boolean skipLocking) {
60-
super(context, config, table, instantTime, commitInstant, deleteInstants, skipTimelinePublish, useMarkerBasedStrategy, skipLocking);
59+
super(context, config, table, instantTime, commitInstant, deleteInstants, skipTimelinePublish, skipLocking);
6160
}
6261

6362
@Override

‎hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -587,4 +587,59 @@ public void testRollbackWithRequestedRollbackPlan(boolean enableMetadataTable, b
587587
}
588588
}
589589
}
590+
591+
@Test
592+
public void testFallbackToListingBasedRollbackForCompletedInstant() throws Exception {
593+
// Let's create some commit files and base files
594+
final String p1 = "2016/05/01";
595+
final String p2 = "2016/05/02";
596+
final String p3 = "2016/05/06";
597+
final String commitTime1 = "20160501010101";
598+
final String commitTime2 = "20160502020601";
599+
final String commitTime3 = "20160506030611";
600+
Map<String, String> partitionAndFileId1 = new HashMap<String, String>() {
601+
{
602+
put(p1, "id11");
603+
put(p2, "id12");
604+
put(p3, "id13");
605+
}
606+
};
607+
Map<String, String> partitionAndFileId2 = new HashMap<String, String>() {
608+
{
609+
put(p1, "id21");
610+
put(p2, "id22");
611+
put(p3, "id23");
612+
}
613+
};
614+
Map<String, String> partitionAndFileId3 = new HashMap<String, String>() {
615+
{
616+
put(p1, "id31");
617+
put(p2, "id32");
618+
put(p3, "id33");
619+
}
620+
};
621+
622+
HoodieWriteConfig config = HoodieWriteConfig.newBuilder().withPath(basePath)
623+
.withRollbackUsingMarkers(true) // rollback using markers to test fallback to listing based rollback for completed instant
624+
.withCleanConfig(HoodieCleanConfig.newBuilder()
625+
.withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY).build())
626+
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build()).build();
627+
628+
// create test table with all commits completed
629+
HoodieTestTable testTable = HoodieMetadataTestTable.of(metaClient, SparkHoodieBackedTableMetadataWriter.create(metaClient.getHadoopConf(), config, context));
630+
testTable.withPartitionMetaFiles(p1, p2, p3)
631+
.addCommit(commitTime1)
632+
.withBaseFilesInPartitions(partitionAndFileId1)
633+
.addCommit(commitTime2)
634+
.withBaseFilesInPartitions(partitionAndFileId2)
635+
.addCommit(commitTime3)
636+
.withBaseFilesInPartitions(partitionAndFileId3);
637+
638+
try (SparkRDDWriteClient client = getHoodieWriteClient(config)) {
639+
client.rollback(commitTime3);
640+
assertFalse(testTable.inflightCommitExists(commitTime3));
641+
assertFalse(testTable.baseFilesExist(partitionAndFileId3, commitTime3));
642+
assertTrue(testTable.baseFilesExist(partitionAndFileId2, commitTime2));
643+
}
644+
}
590645
}

‎hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java‎

Lines changed: 3 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2297,20 +2297,9 @@ private void testRollbackAfterConsistencyCheckFailureUsingFileList(boolean rollb
22972297
"With optimistic CG, first commit should succeed. commit file should be present");
22982298
// Marker directory must be removed after rollback
22992299
assertFalse(metaClient.getFs().exists(new Path(metaClient.getMarkerFolderPath(instantTime))));
2300-
if (rollbackUsingMarkers) {
2301-
// rollback of a completed commit should fail if marked based rollback is used.
2302-
try {
2303-
client.rollback(instantTime);
2304-
fail("Rollback of completed commit should throw exception");
2305-
} catch (HoodieRollbackException e) {
2306-
// ignore
2307-
}
2308-
} else {
2309-
// rollback of a completed commit should succeed if using list based rollback
2310-
client.rollback(instantTime);
2311-
assertFalse(testTable.commitExists(instantTime),
2312-
"After explicit rollback, commit file should not be present");
2313-
}
2300+
client.rollback(instantTime);
2301+
assertFalse(testTable.commitExists(instantTime),
2302+
"After explicit rollback, commit file should not be present");
23142303
}
23152304
}
23162305

‎hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java‎

Lines changed: 2 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -45,9 +45,9 @@
4545
import org.apache.hudi.table.HoodieTable;
4646
import org.apache.hudi.table.marker.WriteMarkersFactory;
4747
import org.apache.hudi.testutils.MetadataMergeWriteStatus;
48+
4849
import org.apache.spark.api.java.JavaRDD;
4950
import org.junit.jupiter.api.AfterEach;
50-
import org.junit.jupiter.api.Assertions;
5151
import org.junit.jupiter.api.BeforeEach;
5252
import org.junit.jupiter.api.Test;
5353
import org.junit.jupiter.params.ParameterizedTest;
@@ -89,7 +89,7 @@ public void tearDown() throws Exception {
8989
}
9090

9191
@ParameterizedTest
92-
@ValueSource(booleans = {true, false})
92+
@ValueSource(booleans = {true})
9393
public void testMergeOnReadRollbackActionExecutor(boolean isUsingMarkers) throws IOException {
9494
//1. prepare data and assert data result
9595
List<FileSlice> firstPartitionCommit2FileSlices = new ArrayList<>();
@@ -281,21 +281,6 @@ public void testRollbackForCanIndexLogFile() throws IOException {
281281
assertEquals(1, partitionMetadata.getSuccessDeleteFiles().size());
282282
}
283283

284-
@Test
285-
public void testFailForCompletedInstants() {
286-
Assertions.assertThrows(IllegalArgumentException.class, () -> {
287-
HoodieInstant rollBackInstant = new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "002");
288-
new MergeOnReadRollbackActionExecutor(context, getConfigBuilder().build(),
289-
getHoodieTable(metaClient, getConfigBuilder().build()),
290-
"003",
291-
rollBackInstant,
292-
true,
293-
true,
294-
true,
295-
false).execute();
296-
});
297-
}
298-
299284
/**
300285
* Test Cases for rolling back when there is no base file.
301286
*/

0 commit comments

Comments
 (0)