[jira] [Created] (FLINK-28941) Savepoint ignores MaxConcurrentCheckpoint limit in aligned checkpoint case

["Yunfeng Zhou (Jira)" <ji...@apache.org>]

Yunfeng Zhou created FLINK-28941:
------------------------------------

             Summary: Savepoint ignores MaxConcurrentCheckpoint limit in aligned checkpoint case
                 Key: FLINK-28941
                 URL: https://issues.apache.org/jira/browse/FLINK-28941
             Project: Flink
          Issue Type: Bug
          Components: Runtime / Checkpointing
    Affects Versions: 1.16.0
            Reporter: Yunfeng Zhou


When the unaligned checkpoint is disabled, savepoints would be set as forced[1], which means they can ignore the maxConcurrentCheckpoint limit[2] and lead to the situation that there are more than maxConcurrentCheckpoint running simultaneously. 

This behavior is incompatible with OperatorCoordinatorHolder, which requires that there should be at most one pending checkpoint at a time. As a result, exceptions, as follows, might be thrown[3].


{code:java}
java.lang.IllegalStateException: Cannot mark for checkpoint 9, already marked for checkpoint 8
	at org.apache.flink.runtime.operators.coordination.SubtaskGatewayImpl.markForCheckpoint(SubtaskGatewayImpl.java:185) ~[flink-runtime-1.16-SNAPSHOT.jar:1.16-SNAPSHOT]
	at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.lambda$checkpointCoordinatorInternal$6(OperatorCoordinatorHolder.java:328) ~[flink-runtime-1.16-SNAPSHOT.jar:1.16-SNAPSHOT]
	at java.util.HashMap.forEach(HashMap.java:1289) ~[?:1.8.0_292]
	at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.checkpointCoordinatorInternal(OperatorCoordinatorHolder.java:327) ~[flink-runtime-1.16-SNAPSHOT.jar:1.16-SNAPSHOT]
	at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.lambda$checkpointCoordinator$0(OperatorCoordinatorHolder.java:243) ~[flink-runtime-1.16-SNAPSHOT.jar:1.16-SNAPSHOT]
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRunAsync$4(AkkaRpcActor.java:453) ~[classes/:?]
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68) ~[classes/:?]
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:453) ~[classes/:?]
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:218) ~[classes/:?]
	at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:84) ~[classes/:?]
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:168) ~[classes/:?]
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at scala.PartialFunction.applyOrElse(PartialFunction.scala:123) [scala-library-2.12.7.jar:?]
	at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122) [scala-library-2.12.7.jar:?]
	at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) [scala-library-2.12.7.jar:?]
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172) [scala-library-2.12.7.jar:?]
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172) [scala-library-2.12.7.jar:?]
	at akka.actor.Actor.aroundReceive(Actor.scala:537) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at akka.actor.Actor.aroundReceive$(Actor.scala:535) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at akka.actor.ActorCell.invoke(ActorCell.scala:548) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at akka.dispatch.Mailbox.run(Mailbox.scala:231) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at akka.dispatch.Mailbox.exec(Mailbox.scala:243) [akka-actor_2.12-2.6.15.jar:2.6.15]
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) [?:1.8.0_292]
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) [?:1.8.0_292]
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) [?:1.8.0_292]
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175) [?:1.8.0_292]
{code}


[1] https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDecider.java#L160-L164
[2] https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java#L444-L449
[3] https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=39860&view=logs&j=219f6d90-20a2-5863-7c1b-c80377a1018f&t=20186858-1485-5059-c9c6-446952519524&s=ab6e269b-88b2-5ded-2544-4aa5b1124530



--
This message was sent by Atlassian Jira
(v8.20.10#820010)