[HUDI-6151] Rollback previously applied commits to MDT when operations are retried. #8604

prashantwason · 2023-04-28T22:26:13Z

[HUDI-6151] Rollback previously applied commits to MDT when operations are retried.

Change Logs

Operations like Clean, Compaction are retried after failures with the same instant time. If the previous run of the operation successfully committed to the MDT but failed to commit to the dataset, then the operation will be retried later with the same instantTime causing duplicate updates applied to MDT.

Currently, we simply delete the completed deltacommit without rolling back the deltacommit.

To handle this, we detect a replay of operation and rollback any changes from that operation in MDT.

Impact

Fixes the issue of duplicate log blocks written in the MDT. This is deterimental for indexes where duplicates are not allowed.

Risk level (write none, low medium or high below)

None. Unit test has been added.

Documentation Update

None

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable
CI passed

danny0405 · 2023-04-30T03:28:06Z

...park-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java

        // if this is a new commit being applied to metadata for the first time
-        writeClient.startCommitWithTime(instantTime);
+        LOG.info("New commit at " + instantTime + " being applied to MDT");
      } else {


to MDT -> to MDT.

Can we sync the logic also to FlinkHoodieBackedTableMetadataWriter ?

...park-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java

nsivabalan · 2023-04-30T17:28:14Z

...park-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java

+            alreadyCompletedInstant.isPresent() ? "Already" : "Partially", instantTime));
+
+        // Rollback the previous committed commit
+        if (!writeClient.rollback(instantTime)) {


Trying to gauge if we really need this.

I guess in next couple of patches, you are going to add below change:

Any rollback in DT will be an actual rollback in MDT as well.

having said that, lets go through this use-case.

Compaction Commit C5 is inflight in DT and succeeded in MDT, but crashed in DT.
so on restart, a rollback is triggered in DT. which when gets into MDT territory, will rollback the succeeded commit in MDT. So, it will be automatically taken care of.

After rollback of C5 is completed, C5 will be re-attempted in DT. and when it gets into MDT territory, there won't be any traces of DC5 at all. So, wondering when exactly we will hit this case?

if we are talking about a partially failed commit in MDT:

Compaction Commit C5 is inflight in DT and DC5 in MDT is also partitally committed and crashed.
On restart, any new operation in DT when it gets into MDT territory, on deducting a partial commit in MDT, a rollback will be triggered eagerly. Ref:

hudi/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java

Line 151 in 04e54a6

if (dataWriteConfig.getFailedWritesCleanPolicy().isEager()

So, this case is also taken care of.

will rollback the succeeded commit in MDT. So, it will be automatically taken care of.

The current code on master just removes the .complete metadata file, is that a rollback you are mentioning about？ To keep sync with regular rollback, I think triggering a real rollback action is necessary.

Older solution of removing the completed action and reattempt won't work in all scenarios. We will have to consider the following scenarios:
(1) c1.commit failed on the main dataset; On MDT, c1.deltacommit was completed.
(a) with record index enabled, new log block was added to the log file by c1.deltacommit. Simply removing deltacommit, may not be enough and will require additional action to rollback the logblock, to keep the log file consistent.
(2) c1.clean was attempted. c1.deltacommit was completed. When clean is retried, second attempt could bring in some of the files that were in the "failed" list of the first attempt (vs the "success" list).
(3) c1.rollback was attempted. c1.deltacommit was completed. (We fixed an issue with incomplete rollback, with MDT updated with deltacommit, scenario. This change played a role in this scenario as well).

Yeah, fix the rollback in sync with normal DT can avoid many potential bugs, +1 for this direction.

nbalajee · 2023-05-04T21:07:42Z

LGTM.

nbalajee

LGTM

danny0405 · 2023-05-05T03:56:30Z

+1, we are good if the test failures are resolved.

nsivabalan · 2023-05-08T23:49:47Z

once the CI is green, we can land

danny0405 · 2023-05-09T11:43:06Z