[HUDI-8808] Fix concurrent execution of appending rollback blocks in the same file group

[yihua]

Change Logs

When there are multiple log files generated in the same file group in a inflight deltacommit, rollback of such a deltacommit can fail, because (1) the rollback plan contains multiple rollback requests targeting the log files to roll back in the same file group (2) concurrent execution of these rollback requests in Spark executors cause creation of new rollback log files (appending rollback blocks) in parallel, which involves determining the new log version concurrently, leading to the same new log version to be used in multiple executors, causing marker creation to fail (i.e., multiple executors try to create the marker on the same file name, and subsequent marker creation requests fail due to the fact that marker already exists).

Note that this problem only happens for table version 6 and below, and backwards compatible writer in Hudi 1.0 writing table version 6. Hudi 1.x does not append rollback blocks any more because of the new format spec.

This issue can be reproduced by the new test added TestMarkerBasedRollbackStrategy#testRollbackMultipleLogFilesInOneFileGroupInMOR before applying the fix (need to change line assertEquals(1, rollbackRequests.size()); to assertEquals(199, rollbackRequests.size());):

org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 1.0 failed 1 times, most recent failure: Lost task 3.0 in stage 1.0 (TID 103) (192.168.1.16 executor driver): org.apache.hudi.exception.HoodieIOException: [timeline-server-based] Failed to create marker for partition partA, fileName .a1e58d70-ce17-400f-9bfe-f394546f71e2_003.log.201_1-0-1 with IOType APPEND
	at org.apache.hudi.table.marker.TimelineServerBasedWriteMarkers.create(TimelineServerBasedWriteMarkers.java:159)
	at org.apache.hudi.table.marker.WriteMarkers.createIfNotExists(WriteMarkers.java:135)
	at org.apache.hudi.table.action.rollback.BaseRollbackHelper$1.createAppendMarker(BaseRollbackHelper.java:250)
	at org.apache.hudi.table.action.rollback.BaseRollbackHelper$1.preLogFileCreate(BaseRollbackHelper.java:246)
	at org.apache.hudi.common.table.log.HoodieLogFormatWriter.createNewFile(HoodieLogFormatWriter.java:241)
	at org.apache.hudi.common.table.log.HoodieLogFormatWriter.getOutputStream(HoodieLogFormatWriter.java:128)
	at org.apache.hudi.common.table.log.HoodieLogFormatWriter.appendBlocks(HoodieLogFormatWriter.java:152)
	at org.apache.hudi.common.table.log.HoodieLogFormatWriter.appendBlock(HoodieLogFormatWriter.java:143)
	at org.apache.hudi.table.action.rollback.BaseRollbackHelper.lambda$maybeDeleteAndCollectStats$b2977713$1(BaseRollbackHelper.java:180)
	at org.apache.hudi.client.common.HoodieSparkEngineContext.lambda$flatMap$7d470b86$1(HoodieSparkEngineContext.java:150)

This PR makes two fixes to tackle the problem:
(1) For rollback plan generation, group the log files to roll back based on the file group (partition + file group ID) so that the log files in the same file group only appears in the same rollback request in the rollback plan (see MarkerBasedRollbackStrategy#getRollbackRequests). Note that we only fix marker-based rollback as that's the only supported rollback mode going forward;
(2) There can still be a case that the existing rollback plan generated on the timeline contains multiple rollback requests targeting the log files in the same file group. To avoid marker creation failure from concurrent execution on these original rollback requests, the rollback execution first groups the rollback requests based on the file group (partition + file group ID) in the rollback plan before parallelization (see BaseRollbackHelper#maybeDeleteAndCollectStats).

This PR adds new tests to comprehensively cover the rollback logic (see TestBaseRollbackHelper, TestMarkerBasedRollbackStrategy, TestRollbackUtils, TestMarkerBasedRollbackStrategy).

Impact

Fixes rollback failures due to concurrent execution of appending rollback blocks in the same file group.

Risk level

low

Documentation Update

N/A

Contributor's checklist

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

[hudi-bot]

CI report:

• 118ab64 Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build