You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by co...@apache.org on 2022/03/31 11:56:04 UTC
[hudi] branch master updated: [HUDI-3732] Fixing rollback validation (#5157)
This is an automated email from the ASF dual-hosted git repository.
codope 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 73a21092 [HUDI-3732] Fixing rollback validation (#5157)
73a21092 is described below
commit 73a21092f8f708d6c90ff2ab95b677b58237be9a
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Thu Mar 31 04:55:24 2022 -0700
[HUDI-3732] Fixing rollback validation (#5157)
* Fixing rollback validation
* Adding tests
---
.../action/restore/BaseRestoreActionExecutor.java | 3 ++
.../rollback/BaseRollbackActionExecutor.java | 7 ++++-
.../TestHoodieClientOnCopyOnWriteStorage.java | 35 ++++++++++++++++++++++
.../table/timeline/HoodieActiveTimeline.java | 5 ++++
4 files changed, 49 insertions(+), 1 deletion(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/BaseRestoreActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/BaseRestoreActionExecutor.java
index 9025623..1fac279 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/BaseRestoreActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/BaseRestoreActionExecutor.java
@@ -136,6 +136,9 @@ public abstract class BaseRestoreActionExecutor<T extends HoodieRecordPayload, I
.filter(instant -> HoodieActiveTimeline.GREATER_THAN.test(instant.getTimestamp(), restoreInstantTime))
.collect(Collectors.toList());
instantsToRollback.forEach(entry -> {
+ if (entry.isCompleted()) {
+ table.getActiveTimeline().deleteCompletedRollback(entry);
+ }
table.getActiveTimeline().deletePending(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.ROLLBACK_ACTION, entry.getTimestamp()));
table.getActiveTimeline().deletePending(new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.ROLLBACK_ACTION, entry.getTimestamp()));
});
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
index d44ba55..8e34f0f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
@@ -186,7 +186,12 @@ public abstract class BaseRollbackActionExecutor<T extends HoodieRecordPayload,
}
}
- List<String> inflights = inflightAndRequestedCommitTimeline.getInstants().map(HoodieInstant::getTimestamp)
+ List<String> inflights = inflightAndRequestedCommitTimeline.getInstants().filter(instant -> {
+ if (!instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
+ return true;
+ }
+ return !ClusteringUtils.isPendingClusteringInstant(table.getMetaClient(), instant);
+ }).map(HoodieInstant::getTimestamp)
.collect(Collectors.toList());
if ((instantTimeToRollback != null) && !inflights.isEmpty()
&& (inflights.indexOf(instantTimeToRollback) != inflights.size() - 1)) {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
index 3b78954..a6a3703 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
@@ -1399,6 +1399,41 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
testInsertAndClustering(clusteringConfig, populateMetaFields, true, false, SqlQueryEqualityPreCommitValidator.class.getName(), COUNT_SQL_QUERY_FOR_VALIDATION, "");
}
+ @Test
+ public void testRolblackOfRegularCommitWithPendingReplaceCommitInTimeline() throws Exception {
+ HoodieClusteringConfig clusteringConfig = HoodieClusteringConfig.newBuilder().withClusteringMaxNumGroups(10)
+ .withClusteringTargetPartitions(0).withInlineClusteringNumCommits(1).withInlineClustering(true)
+ .withPreserveHoodieCommitMetadata(true).build();
+ // trigger clustering, but do not complete
+ testInsertAndClustering(clusteringConfig, true, false, false, SqlQueryEqualityPreCommitValidator.class.getName(), COUNT_SQL_QUERY_FOR_VALIDATION, "");
+
+ // trigger another partial commit, followed by valid commit. rollback of partial commit should succeed.
+ HoodieWriteConfig.Builder cfgBuilder = getConfigBuilder().withAutoCommit(false);
+ SparkRDDWriteClient client = getHoodieWriteClient(cfgBuilder.build());
+ String commitTime1 = HoodieActiveTimeline.createNewInstantTime();
+ List<HoodieRecord> records1 = dataGen.generateInserts(commitTime1, 200);
+ client.startCommitWithTime(commitTime1);
+ JavaRDD<HoodieRecord> insertRecordsRDD1 = jsc.parallelize(records1, 2);
+ JavaRDD<WriteStatus> statuses = client.upsert(insertRecordsRDD1, commitTime1);
+ List<WriteStatus> statusList = statuses.collect();
+ assertNoWriteErrors(statusList);
+
+ HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(basePath).build();
+ assertEquals(2, metaClient.getActiveTimeline().getCommitsTimeline().filterInflightsAndRequested().countInstants());
+
+ // trigger another commit. this should rollback latest partial commit.
+ records1 = dataGen.generateInserts(commitTime1, 200);
+ client.startCommitWithTime(commitTime1);
+ insertRecordsRDD1 = jsc.parallelize(records1, 2);
+ statuses = client.upsert(insertRecordsRDD1, commitTime1);
+ statusList = statuses.collect();
+ assertNoWriteErrors(statusList);
+ client.commit(commitTime1, statuses);
+ metaClient.reloadActiveTimeline();
+ // rollback should have succeeded. Essentially, the pending clustering should not hinder the rollback of regular commits.
+ assertEquals(1, metaClient.getActiveTimeline().getCommitsTimeline().filterInflightsAndRequested().countInstants());
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testInlineScheduleClustering(boolean scheduleInlineClustering) throws IOException {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
index 36dd536..f7dc2f6 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
@@ -201,6 +201,11 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
deleteInstantFile(instant);
}
+ public void deleteCompletedRollback(HoodieInstant instant) {
+ ValidationUtils.checkArgument(instant.isCompleted());
+ deleteInstantFile(instant);
+ }
+
public static void deleteInstantFile(FileSystem fs, String metaPath, HoodieInstant instant) {
try {
fs.delete(new Path(metaPath, instant.getFileName()), false);