Hi, Maybe Chesney you are right, but I’m not sure. TwoPhaseCommitSink was based on Pravega’s sink for Flink, which was implemented by Stephan, and it has the same logic [1]. If I remember the discussions with Stephan/Till, the way how Flink is using Akka probably guarantees that messages will be always delivered, except of some failure, so `notifyCheckpointComplete` could be missed probably only if a failure happens between snapshot and arrival of the notification. Receiving the same notification twice should be impossible (based on the knowledge passed to me from Till/Stephan).
However, for one thing, if that’s possible, then the code should adjusted accordingly. On the other hand, maybe there is no harm in relaxing the contract? Even if we miss this notification (because of some re-ordering?), next one will subsume the missed one and commit everything. Piotrek [1] https://github.com/pravega/flink-connectors/blob/master/src/main/java/io/pravega/connectors/flink/FlinkPravegaWriter.java#L567 <https://github.com/pravega/flink-connectors/blob/master/src/main/java/io/pravega/connectors/flink/FlinkPravegaWriter.java#L567> > On 27 Nov 2019, at 13:02, Chesnay Schepler <ches...@apache.org> wrote: > > This looks to me like the TwoPhaseCommitSinkFunction is a bit too strict. The > notification for complete checkpoints is not reliable; it may be late, not > come at all, possibly even in different order than expected. > > As such, if you a simple case of snapshot -> snapshot -> notify -> notify the > sink will always fail with an exception. > > What it should do imo is either a) don't check that there is a pending > transaction or b) track the highest checkpoint id received and optionally > don't fail if the notification is for an older CP. > > @piotr WDYT? > > On 27/11/2019 08:59, Tony Wei wrote: >> Hi, >> >> As the follow up, it seem that savepoint can't be subsumed, so that its >> notification could still be send to each TMs. >> Is this a bug that need to be fixed in TwoPhaseCommitSinkFunction? >> >> Best, >> Tony Wei >> >> Tony Wei <tony19920...@gmail.com <mailto:tony19920...@gmail.com>> 於 >> 2019年11月27日 週三 下午3:43寫道： >> Hi, >> >> I want to raise this question again, since I have had this exception on my >> production job. >> >> The exception is as follows >> >> 2019-11-27 14:47:29 >> >> java.lang.RuntimeException: Error while confirming checkpoint >> at org.apache.flink.runtime.taskmanager.Task$2.run(Task.java:1205) >> at >> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) >> at java.util.concurrent.FutureTask.run(FutureTask.java:266) >> 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) >> Caused by: java.lang.IllegalStateException: checkpoint completed, but no >> transaction pending >> at org.apache.flink.util.Preconditions.checkState(Preconditions.java:195) >> at >> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.notifyCheckpointComplete(TwoPhaseCommitSinkFunction.java:267) >> at >> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.notifyCheckpointComplete(AbstractUdfStreamOperator.java:130) >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask.notifyCheckpointComplete(StreamTask.java:822) >> at org.apache.flink.runtime.taskmanager.Task$2.run(Task.java:1200) >> ... 5 more >> >> And these are the checkpoint / savepoint before the job failed. >> <checkoint.png> >> >> It seems that checkpoint # 675's notification handled the savepoint # 674's >> pending transaction holder, but savepoint #674's notification didn't be >> subsumed or be ignored by JM. >> Therefore, during the checkpoint #676, some tasks got notification before >> getting the checkpoint barrier and led to this exception happened, because >> there was no pending transaction in queue. >> >> Does anyone know the details about subsumed notifications mechanism and how >> checkpoint coordinator handle this situation? Please correct me if I'm >> wrong. Thanks. >> >> Best, >> Tony Wei >> >> Stefan Richter <s.rich...@data-artisans.com >> <mailto:s.rich...@data-artisans.com>> 於 2018年10月8日 週一 下午5:03寫道： >> Hi Pedro, >> >> unfortunately the interesting parts are all removed from the log, we already >> know about the exception itself. In particular, what I would like to see is >> what checkpoints have been triggered and completed before the exception >> happens. >> >> Best, >> Stefan >> >> > Am 08.10.2018 um 10:23 schrieb PedroMrChaves <pedro.mr.cha...@gmail.com >> > <mailto:pedro.mr.cha...@gmail.com>>: >> > >> > Hello, >> > >> > Find attached the jobmanager.log. I've omitted the log lines from other >> > runs, only left the job manager info and the run with the error. >> > >> > jobmanager.log >> > <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/file/t612/jobmanager.log >> > >> > <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/file/t612/jobmanager.log>> >> > >> > >> > >> > >> > Thanks again for your help. >> > >> > Regards, >> > Pedro. >> > >> > >> > >> > ----- >> > Best Regards, >> > Pedro Chaves >> > -- >> > Sent from: >> > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/ >> > <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/> >> >
