[
https://issues.apache.org/jira/browse/HUDI-2841?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17448258#comment-17448258
]
sivabalan narayanan edited comment on HUDI-2841 at 11/23/21, 9:55 PM:
----------------------------------------------------------------------
Need to understand the purpose of "Written log files" in rollback commit
metadata. it does not represent just the log files written by commit of
interest.
Rollback actually logs(written log files) other log files written by future
delta commits. I retried the same scenario w/ non-partitioned dataset so that
log files are part of commit that got failed.
partition1/
baseFile1 // by DC1
logfile1 // by DC2
logfile2 // by DC3 failed mid way.
restart delta streamer.
DC4 creates logFile3. and completes.
and then rollbacks DC3(rollback happens as part of cleaner with lazy policy).
but list of files to be deleted has logFile1, logFile2 and logFile3.
Excerpts from logs:
.hoodie just after DC3 failure.
{code:java}
ls -ltra /tmp/hudi-deltastreamer-ny/ | grep -v crc
total 920
drwxrwxrwt 52 root wheel 1664 Nov 23 14:46 ..
-rw-r--r-- 1 nsb wheel 96 Nov 23 14:46 .hoodie_partition_metadata
-rw-r--r-- 1 nsb wheel 439457 Nov 23 14:46
57993dad-822c-4b28-a9da-2ff6d61b10d0-0_0-23-80_20211123144608551.parquet
-rw-r--r-- 1 nsb wheel 2507 Nov 23 14:46
.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.1_0-63-174
drwxr-xr-x 24 nsb wheel 768 Nov 23 14:47 .hoodie
drwxr-xr-x 11 nsb wheel 352 Nov 23 14:47 .
-rw-r--r-- 1 nsb wheel 2507 Nov 23 14:47
.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.2_0-103-268 {code}
// log.2_ is written by failed commit.
Rollback commit metadata
{code:java}
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: BaseRollbackAction
Executor. rolling back
20211123144713094
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: for partition
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: success delete files 0
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: failed delete files 0
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: rollback log files 1
21/11/23 14:49:15 WARN BaseRollbackActionExecutor:
file:/tmp/hudi-deltastreamer-ny/.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.4_1-0-1
-> 105
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: written log files 3
21/11/23 14:49:15 WARN BaseRollbackActionExecutor:
file:/tmp/hudi-deltastreamer-ny/.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.3_0-26-90
-> 2507
21/11/23 14:49:15 WARN BaseRollbackActionExecutor:
file:/tmp/hudi-deltastreamer-ny/.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.1_0-63-174
-> 2507
21/11/23 14:49:15 WARN BaseRollbackActionExecutor:
file:/tmp/hudi-deltastreamer-ny/.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.2_0-103-268
-> 2507 {code}
// log.3_ is written by DC4.
was (Author: shivnarayan):
There could be even a bigger problem here. Rollback actually rollsback other
log files written by future delta commits. I retried the same scenario w/
non-partitioned dataset so that log files are part of commit that got failed.
partition1/
baseFile1 // by DC1
logfile1 // by DC2
logfile2 // by DC3 failed mid way.
restart delta streamer.
DC4 creates logFile3. and completes.
and then rollbacks DC3(rollback happens as part of cleaner with lazy policy).
but list of files to be deleted has logFile1, logFile2 and logFile3.
Excerpts from logs:
.hoodie just after DC3 failure.
{code:java}
ls -ltra /tmp/hudi-deltastreamer-ny/ | grep -v crc
total 920
drwxrwxrwt 52 root wheel 1664 Nov 23 14:46 ..
-rw-r--r-- 1 nsb wheel 96 Nov 23 14:46 .hoodie_partition_metadata
-rw-r--r-- 1 nsb wheel 439457 Nov 23 14:46
57993dad-822c-4b28-a9da-2ff6d61b10d0-0_0-23-80_20211123144608551.parquet
-rw-r--r-- 1 nsb wheel 2507 Nov 23 14:46
.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.1_0-63-174
drwxr-xr-x 24 nsb wheel 768 Nov 23 14:47 .hoodie
drwxr-xr-x 11 nsb wheel 352 Nov 23 14:47 .
-rw-r--r-- 1 nsb wheel 2507 Nov 23 14:47
.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.2_0-103-268 {code}
// log.2_ is written by failed commit.
Rollback commit metadata
{code:java}
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: BaseRollbackAction
Executor. rolling back
20211123144713094
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: for partition
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: success delete files 0
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: failed delete files 0
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: rollback log files 1
21/11/23 14:49:15 WARN BaseRollbackActionExecutor:
file:/tmp/hudi-deltastreamer-ny/.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.4_1-0-1
-> 105
21/11/23 14:49:15 WARN BaseRollbackActionExecutor: written log files 3
21/11/23 14:49:15 WARN BaseRollbackActionExecutor:
file:/tmp/hudi-deltastreamer-ny/.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.3_0-26-90
-> 2507
21/11/23 14:49:15 WARN BaseRollbackActionExecutor:
file:/tmp/hudi-deltastreamer-ny/.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.1_0-63-174
-> 2507
21/11/23 14:49:15 WARN BaseRollbackActionExecutor:
file:/tmp/hudi-deltastreamer-ny/.57993dad-822c-4b28-a9da-2ff6d61b10d0-0_20211123144608551.log.2_0-103-268
-> 2507 {code}
// log.3_ is written by DC4.
> Rollback of a failed delta commit w/ multi-writer fails
> -------------------------------------------------------
>
> Key: HUDI-2841
> URL: https://issues.apache.org/jira/browse/HUDI-2841
> Project: Apache Hudi
> Issue Type: Bug
> Affects Versions: 0.10.0
> Reporter: sivabalan narayanan
> Priority: Blocker
> Fix For: 0.10.0
>
>
> In multi-writer set up, cleaner policy has to be set to lazy. And failed
> rollbacks are operated upon(actually being rolledback) only by cleaner and
> not eagerly.
> List based rollback for MOR fails with validation that files to be rolledback
> has large timestamp compared to commit being rolledback.
>
> Lets say, timeline is as follows:
> DC1, DC2... and DC3 failed midway.
> Partition1/
> baseFile1 (DC1)
> baseFile2 (DC2) // due to small file handling
> baseFile3 (DC3)
>
> Restarted deltastremaer.
> which does DC4
>
> Partition1/
> baseFile1 (DC1)
> baseFile2 (DC2) // due to small file handling
> baseFile3 (DC3)
> baseFile4 (DC4)
>
> At the end of the commit, cleaner kicks in and tries to rollbacking any
> failed commits. In RollbackUtils, where we fetch the latestFileSlice and find
> all files to be deleted, we have a validation which checks that base commit
> time for every file (in latest file slice) should have commit time < commit
> being rolledback. In this case, baseFile4 has higher timestamp compared to
> DC3 and hence validation fails.
>
> {code:java}
> 21/11/23 09:53:49 DEBUG wire: http-outgoing-11 >> "[\r][\n]"
> 21/11/23 09:53:49 ERROR Executor: Exception in task 32.0 in stage 45.0 (TID
> 3191)
> java.lang.IllegalArgumentException
> at
> org.apache.hudi.common.util.ValidationUtils.checkArgument(ValidationUtils.java:31)
> at
> org.apache.hudi.table.action.rollback.RollbackUtils.lambda$generateAppendRollbackBlocksAction$0(RollbackUtils.java:254)
> at
> java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:174)
> at
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
> at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
> at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
> at
> java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
> at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
> at
> java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
> at
> org.apache.hudi.table.action.rollback.RollbackUtils.generateAppendRollbackBlocksAction(RollbackUtils.java:266)
> at
> org.apache.hudi.table.action.rollback.RollbackUtils.lambda$generateRollbackRequestsUsingFileListingMOR$e97f040e$1(RollbackUtils.java:210)
> at
> org.apache.hudi.client.common.HoodieSparkEngineContext.lambda$flatMap$7d470b86$1(HoodieSparkEngineContext.java:134)
> at
> org.apache.spark.api.java.JavaRDDLike.$anonfun$flatMap$1(JavaRDDLike.scala:125)
> at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)
> at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:490)
> at scala.collection.Iterator.foreach(Iterator.scala:941)
> at scala.collection.Iterator.foreach$(Iterator.scala:941)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
> at scala.collection.generic.Growable.$plus$plus$eq(Growable.scala:62)
> at scala.collection.generic.Growable.$plus$plus$eq$(Growable.scala:53)
> at
> scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:105)
> at
> scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:49)
> at scala.collection.TraversableOnce.to(TraversableOnce.scala:315)
> at scala.collection.TraversableOnce.to$(TraversableOnce.scala:313)
> at scala.collection.AbstractIterator.to(Iterator.scala:1429)
> at
> scala.collection.TraversableOnce.toBuffer(TraversableOnce.scala:307)
> at
> scala.collection.TraversableOnce.toBuffer$(TraversableOnce.scala:307)
> at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1429)
> at scala.collection.TraversableOnce.toArray(TraversableOnce.scala:294)
> at
> scala.collection.TraversableOnce.toArray$(TraversableOnce.scala:288)
> at scala.collection.AbstractIterator.toArray(Iterator.scala:1429)
> at org.apache.spark.rdd.RDD.$anonfun$collect$2(RDD.scala:1004)
> at
> org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2139)
> at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
> at org.apache.spark.scheduler.Task.run(Task.scala:127)
> at
> org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)
> at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> 21/11/23 09:53:49 WARN RollbackUtils: Rollback Instant time :
> 20211123095048787 {code}
>
> Code of interest
> {code:java}
> private static List<ListingBasedRollbackRequest>
> generateAppendRollbackBlocksAction(String partitionPath, HoodieInstant
> rollbackInstant,
> HoodieCommitMetadata commitMetadata, HoodieTable table) {
>
> ValidationUtils.checkArgument(rollbackInstant.getAction().equals(HoodieTimeline.DELTA_COMMIT_ACTION));
> // wStat.getPrevCommit() might not give the right commit time in the
> following
> // scenario : If a compaction was scheduled, the new commitTime associated
> with the requested compaction will be
> // used to write the new log files. In this case, the commit time for the
> log file is the compaction requested time.
> // But the index (global) might store the baseCommit of the base and not
> the requested, hence get the
> // baseCommit always by listing the file slice
> Map<String, String> fileIdToBaseCommitTimeForLogMap =
> table.getSliceView().getLatestFileSlices(partitionPath)
> .collect(Collectors.toMap(FileSlice::getFileId,
> FileSlice::getBaseInstantTime));
> return
> commitMetadata.getPartitionToWriteStats().get(partitionPath).stream().filter(wStat
> -> {
> // Filter out stats without prevCommit since they are all inserts
> boolean validForRollback = (wStat != null) &&
> (!wStat.getPrevCommit().equals(HoodieWriteStat.NULL_COMMIT))
> && (wStat.getPrevCommit() != null) &&
> fileIdToBaseCommitTimeForLogMap.containsKey(wStat.getFileId());
> if (validForRollback) {
> // For sanity, log instant time can never be less than base-commit on
> which we are rolling back
> ValidationUtils
>
> .checkArgument(HoodieTimeline.compareTimestamps(fileIdToBaseCommitTimeForLogMap.get(wStat.getFileId()),
> HoodieTimeline.LESSER_THAN_OR_EQUALS,
> rollbackInstant.getTimestamp()));
> }
> return validForRollback &&
> HoodieTimeline.compareTimestamps(fileIdToBaseCommitTimeForLogMap.get(
> // Base Ts should be strictly less. If equal (for inserts-to-logs),
> the caller employs another option
> // to delete and we should not step on it
> wStat.getFileId()), HoodieTimeline.LESSER_THAN,
> rollbackInstant.getTimestamp());
> }).map(wStat -> {
> String baseCommitTime =
> fileIdToBaseCommitTimeForLogMap.get(wStat.getFileId());
> return
> ListingBasedRollbackRequest.createRollbackRequestWithAppendRollbackBlockAction(partitionPath,
> wStat.getFileId(),
> baseCommitTime);
> }).collect(Collectors.toList());
> } {code}
> Verified with log statements that DC4 base file is what runs into the
> validation error.
>
> But the issue could also happen if incase a compaction had kicked in mid way
> and cleaner policy was very relaxed and so rollback got triggered after 10
> commits or 1 to 2 compactions.
>
> In the code block of interest, we call
> ` table.getSliceView().getLatestFileSlices(partitionPath)`. Instead, I tried
> to use `
> table.getSliceView().getLatestFileSlicesBeforeOrOn(partitionPath,
> rollbackInstant.getTimestamp(), false)`
> But this started including files as part of the next delta commit as well to
> the list of files to be deleted.
>
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
