Flaky test: ITTestHoodieDataSource.testBatchReadWithLimit

[voonhous]

Bug Description

What happened:

What you expected:

Steps to reproduce:
1.
2.
3.

Environment

Hudi version:
Query engine: (Spark/Flink/Trino etc)
Relevant configs:

Logs and Stack Trace

 java.lang.RuntimeException: Failed to fetch next result
      at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:129)
      at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:100)
      at org.apache.flink.table.planner.connectors.CollectDynamicSink$CloseableRowIteratorWrapper.hasNext(CollectDynamicSink.java:247)
      at java.base/java.util.Iterator.forEachRemaining(Iterator.java:132)
      at org.apache.flink.util.CollectionUtil.iterableToList(CollectionUtil.java:119)
      at org.apache.hudi.table.ITTestHoodieDataSource.testBatchReadWithLimit(ITTestHoodieDataSource.java:3250)
      at java.base/java.lang.reflect.Method.invoke(Method.java:566)
      at java.base/java.util.Optional.ifPresent(Optional.java:183)
      at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
      at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
      at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
      at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
      at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
      at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
      at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
      at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
      at java.base/java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:274)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
      at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
      at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
      at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
      at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
      at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
      at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
      at java.base/java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:274)
      at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1655)
      at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
      at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
      at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
      at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
      at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
      at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
      at java.base/java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:274)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
      at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1655)
      at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
      at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
      at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
      at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
      at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
      at org.apache.pekko.actor.Actor.aroundReceive(Actor.scala:547)
      at org.apache.pekko.actor.Actor.aroundReceive$(Actor.scala:545)
      at org.apache.pekko.actor.AbstractActor.aroundReceive(AbstractActor.scala:229)
      at org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590)
      at org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557)
      at org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:272)
      at org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:233)
      at org.apache.pekko.dispatch.Mailbox.exec(Mailbox.scala:245)
      at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
      at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
      at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
      at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
      at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)
  Caused by: java.lang.NullPointerException
      at org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader.readNextRowGroup(ParquetColumnarRowSplitReader.java:334)
      at org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader.nextBatch(ParquetColumnarRowSplitReader.java:319)
      at org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader.ensureBatch(ParquetColumnarRowSplitReader.java:296)
      at org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader.reachedEnd(ParquetColumnarRowSplitReader.java:277)
      at org.apache.hudi.table.format.ParquetSplitRecordIterator.hasNext(ParquetSplitRecordIterator.java:42)
      at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39)
      at org.apache.hudi.common.table.read.HoodieFileGroupReader.hasNext(HoodieFileGroupReader.java:245)
      at org.apache.hudi.common.table.read.HoodieFileGroupReader$HoodieFileGroupReaderIterator.hasNext(HoodieFileGroupReader.java:334)
      at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39)
      at org.apache.hudi.source.reader.BatchRecords.nextRecordFromSplit(BatchRecords.java:81)
      at org.apache.hudi.source.reader.BatchRecords.nextRecordFromSplit(BatchRecords.java:35)
      at org.apache.hudi.source.reader.RecordLimiter$1.nextRecordFromSplit(RecordLimiter.java:65)
      at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:199)
      at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:443)
      at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
      at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:638)
      at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917)
      at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970)
      at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949)
      at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763)
      at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
      at java.base/java.lang.Thread.run(Thread.java:829)

  [INFO]
  [INFO] Results:
  [INFO]
  Error:  Errors:
  Error:    ITTestHoodieDataSource.testBatchReadWithLimit:3250 » Runtime Failed to fetch next result

[HuangZhenQiu]

@voonhous
The test case is added by me. I will work on the fix.

[voonhous]

Oh, thank you! I've only encountered it once or twice, if it's not worth the time to fix it, so be it. Just annotating this down in case i lose context of things.

[voonhous]

Another error:

Error:  Tests run: 241, Failures: 0, Errors: 1, Skipped: 1, Time elapsed: 1181 s <<< FAILURE! -- in org.apache.hudi.table.ITTestHoodieDataSource
Error:  org.apache.hudi.table.ITTestHoodieDataSource.testStreamReadFromSpecifiedCommitWithChangelog(HoodieCDCSupplementalLoggingMode, boolean)[4] -- Time elapsed: 7.034 s <<< ERROR!

Saving the full log in the case where logs get deleted:
https://github.com/apache/hudi/actions/runs/24629579196/job/72014469221?pr=18529

Might be a different cause, but it's flaking on the same test class. The stacktrace from the issue above that i have pasted might not even be the actual root cause.

job-logs.txt

[voonhous]

Not reproducible anymore, will close.