[
https://issues.apache.org/jira/browse/FLINK-35382?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17847783#comment-17847783
]
Jinzhong Li commented on FLINK-35382:
-------------------------------------
1. Because FLINK-32085 has not been resolved, the changelog StateBackend still
do not support snapshot-file-merging(FLIP_306) at this time.
So the FILE_MERGING_ENABLED config should not be enabled for the
ChangelogCompatibilityITCase yet.
2.In FLINK-32092, the FILE_MERGING_ENABLED config is randomly enabled in
TestStreamEnvironment, and try to disable it in ChangelogCompatibilityITCase,
but it does not actually work.
I will fix it by disable FILE_MERGING_ENABLED explicitly in
ChangelogCompatibilityITCase.
> ChangelogCompatibilityITCase.testRestore fails with an NPE
> ----------------------------------------------------------
>
> Key: FLINK-35382
> URL: https://issues.apache.org/jira/browse/FLINK-35382
> Project: Flink
> Issue Type: Bug
> Affects Versions: 1.20.0
> Reporter: Ryan Skraba
> Priority: Critical
> Labels: test-stability
>
> * 1.20 Java 8 / Test (module: tests)
> https://github.com/apache/flink/actions/runs/9110398985/job/25045798401#step:10:8192
> It looks like there can be a [NullPointerException at this
> line|https://github.com/apache/flink/blob/9a5a99b1a30054268bbde36d565cbb1b81018890/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java#L666]
> causing a test failure:
> {code}
> Error: 10:36:23 10:36:23.312 [ERROR] Tests run: 9, Failures: 0, Errors: 1,
> Skipped: 0, Time elapsed: 19.31 s <<< FAILURE! -- in
> org.apache.flink.test.state.ChangelogCompatibilityITCase
> Error: 10:36:23 10:36:23.313 [ERROR]
> org.apache.flink.test.state.ChangelogCompatibilityITCase.testRestore[startWithChangelog=false,
> restoreWithChangelog=true, restoreFrom=CHECKPOINT, allowStore=true,
> allowRestore=true] -- Time elapsed: 1.492 s <<< ERROR!
> May 16 10:36:23 java.lang.RuntimeException:
> org.opentest4j.AssertionFailedError: Graph is in globally terminal state
> (FAILED)
> May 16 10:36:23 at
> org.apache.flink.test.state.ChangelogCompatibilityITCase.tryRun(ChangelogCompatibilityITCase.java:204)
> May 16 10:36:23 at
> org.apache.flink.test.state.ChangelogCompatibilityITCase.restoreAndValidate(ChangelogCompatibilityITCase.java:190)
> May 16 10:36:23 at java.util.Optional.ifPresent(Optional.java:159)
> May 16 10:36:23 at
> org.apache.flink.test.state.ChangelogCompatibilityITCase.testRestore(ChangelogCompatibilityITCase.java:118)
> May 16 10:36:23 at java.lang.reflect.Method.invoke(Method.java:498)
> May 16 10:36:23 Caused by: org.opentest4j.AssertionFailedError: Graph is in
> globally terminal state (FAILED)
> May 16 10:36:23 at
> org.junit.jupiter.api.AssertionUtils.fail(AssertionUtils.java:42)
> May 16 10:36:23 at
> org.junit.jupiter.api.Assertions.fail(Assertions.java:150)
> May 16 10:36:23 at
> org.apache.flink.runtime.testutils.CommonTestUtils.lambda$waitForAllTaskRunning$3(CommonTestUtils.java:214)
> May 16 10:36:23 at
> org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition(CommonTestUtils.java:151)
> May 16 10:36:23 at
> org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition(CommonTestUtils.java:145)
> May 16 10:36:23 at
> org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning(CommonTestUtils.java:209)
> May 16 10:36:23 at
> org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning(CommonTestUtils.java:182)
> May 16 10:36:23 at
> org.apache.flink.test.state.ChangelogCompatibilityITCase.submit(ChangelogCompatibilityITCase.java:284)
> May 16 10:36:23 at
> org.apache.flink.test.state.ChangelogCompatibilityITCase.tryRun(ChangelogCompatibilityITCase.java:197)
> May 16 10:36:23 ... 4 more
> May 16 10:36:23 Caused by: org.apache.flink.runtime.JobException:
> org.apache.flink.runtime.JobException: Recovery is suppressed by
> NoRestartBackoffTimeStrategy
> May 16 10:36:23 at
> org.apache.flink.runtime.executiongraph.failover.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:219)
> May 16 10:36:23 at
> org.apache.flink.runtime.executiongraph.failover.ExecutionFailureHandler.handleFailureAndReport(ExecutionFailureHandler.java:166)
> May 16 10:36:23 at
> org.apache.flink.runtime.executiongraph.failover.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:121)
> May 16 10:36:23 at
> org.apache.flink.runtime.scheduler.DefaultScheduler.recordTaskFailure(DefaultScheduler.java:279)
> May 16 10:36:23 at
> org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:270)
> May 16 10:36:23 at
> org.apache.flink.runtime.scheduler.DefaultScheduler.onTaskFailed(DefaultScheduler.java:263)
> May 16 10:36:23 at
> org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:788)
> May 16 10:36:23 at
> org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:765)
> May 16 10:36:23 at
> org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83)
> May 16 10:36:23 at
> org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:496)
> May 16 10:36:23 at java.lang.reflect.Method.invoke(Method.java:498)
> May 16 10:36:23 at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318)
> May 16 10:36:23 at
> org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
> May 16 10:36:23 at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:316)
> May 16 10:36:23 at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:229)
> May 16 10:36:23 at
> org.apache.flink.runtime.rpc.pekko.FencedPekkoRpcActor.handleRpcMessage(FencedPekkoRpcActor.java:88)
> May 16 10:36:23 at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleMessage(PekkoRpcActor.java:174)
> May 16 10:36:23 at
> org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:33)
> May 16 10:36:23 at
> org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:29)
> May 16 10:36:23 at
> scala.PartialFunction.applyOrElse(PartialFunction.scala:127)
> May 16 10:36:23 at
> scala.PartialFunction.applyOrElse$(PartialFunction.scala:126)
> May 16 10:36:23 at
> org.apache.pekko.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:29)
> May 16 10:36:23 at
> scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175)
> May 16 10:36:23 at
> scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
> May 16 10:36:23 at
> scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
> May 16 10:36:23 at
> org.apache.pekko.actor.Actor.aroundReceive(Actor.scala:547)
> May 16 10:36:23 at
> org.apache.pekko.actor.Actor.aroundReceive$(Actor.scala:545)
> May 16 10:36:23 at
> org.apache.pekko.actor.AbstractActor.aroundReceive(AbstractActor.scala:229)
> May 16 10:36:23 at
> org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590)
> May 16 10:36:23 at
> org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557)
> May 16 10:36:23 at
> org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:280)
> May 16 10:36:23 at
> org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:241)
> May 16 10:36:23 at
> org.apache.pekko.dispatch.Mailbox.exec(Mailbox.scala:253)
> May 16 10:36:23 at
> java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
> May 16 10:36:23 at
> java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
> May 16 10:36:23 at
> java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
> May 16 10:36:23 at
> java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175)
> May 16 10:36:23 Caused by: java.lang.NullPointerException:
> java.lang.NullPointerException
> May 16 10:36:23 at
> org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManagerBase.isManagedByFileMergingManager(FileMergingSnapshotManagerBase.java:666)
> May 16 10:36:23 at
> org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManagerBase.lambda$null$3(FileMergingSnapshotManagerBase.java:620)
> May 16 10:36:23 at java.util.HashMap.computeIfAbsent(HashMap.java:1128)
> May 16 10:36:23 at
> org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManagerBase.lambda$restoreStateHandles$4(FileMergingSnapshotManagerBase.java:616)
> May 16 10:36:23 at
> java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
> May 16 10:36:23 at
> java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
> May 16 10:36:23 at
> java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
> May 16 10:36:23 at
> java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
> May 16 10:36:23 at
> java.util.stream.Streams$StreamBuilderImpl.forEachRemaining(Streams.java:419)
> May 16 10:36:23 at
> java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:647)
> May 16 10:36:23 at
> java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:272)
> May 16 10:36:23 at
> java.util.ArrayList$Itr.forEachRemaining(ArrayList.java:901)
> May 16 10:36:23 at
> java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801)
> May 16 10:36:23 at
> java.util.stream.Streams$ConcatSpliterator.forEachRemaining(Streams.java:742)
> May 16 10:36:23 at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
> May 16 10:36:23 at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
> May 16 10:36:23 at
> java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
> May 16 10:36:23 at
> java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
> May 16 10:36:23 at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
> May 16 10:36:23 at
> java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
> May 16 10:36:23 at
> java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:272)
> May 16 10:36:23 at
> java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
> May 16 10:36:23 at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
> May 16 10:36:23 at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
> May 16 10:36:23 at
> java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
> May 16 10:36:23 at
> java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
> May 16 10:36:23 at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
> May 16 10:36:23 at
> java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
> May 16 10:36:23 at
> org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManagerBase.restoreStateHandles(FileMergingSnapshotManagerBase.java:613)
> May 16 10:36:23 at
> org.apache.flink.runtime.checkpoint.filemerging.SubtaskFileMergingManagerRestoreOperation.restore(SubtaskFileMergingManagerRestoreOperation.java:98)
> May 16 10:36:23 at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.registerRestoredStateToFileMergingManager(StreamTaskStateInitializerImpl.java:353)
> May 16 10:36:23 at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
> May 16 10:36:23 at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:267)
> May 16 10:36:23 at
> org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
> May 16 10:36:23 at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreStateAndGates(StreamTask.java:851)
> May 16 10:36:23 at
> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$restoreInternal$4(StreamTask.java:805)
> May 16 10:36:23 at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
> May 16 10:36:23 at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:805)
> May 16 10:36:23 at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:764)
> May 16 10:36:23 at
> org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:960)
> May 16 10:36:23 at
> org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:929)
> May 16 10:36:23 at
> org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:753)
> May 16 10:36:23 at
> org.apache.flink.runtime.taskmanager.Task.run(Task.java:568)
> May 16 10:36:23 at java.lang.Thread.run(Thread.java:750)
> May 16 10:36:23
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
