Hi Max, I agree that the problem might not be in the acknowledgement itself. A very long checkpoint could go past the subscription acknowledgement deadline (10min is the maximum allowed) and hence the message might be resent yielding the behaviour we see.
In any way, the extreme slow down of checkpoints still remains unexplained. This occurs even if the job simply reads from Pubsub and does nothing else. We do use FsStateBackend using HDFS. The whole setup is deployed in Kubernetes. Any ideas of why this might be happening would be of great help. Thanks, Encho On Mon, Sep 17, 2018 at 4:15 PM Maximilian Michels <[email protected]> wrote: > Hi Encho, > > Thanks for providing more insight into this. I've re-examined the > checkpointing code and couldn't find anything suspicious there. > > > The first job I stopped right when it processed more messages than I > > had loaded. The subscription afterwards had 52 000 unacknowledged > > messages. > > That does sound suspicious with a parallelism of 52, but your other > experiments don't confirm that there is something wrong with the > acknowledgment. Rather, it seems the checkpointing itself is taking > longer and longer. This could also be caused by long acknowlegments, > since this stalls in-progress checkpoints. > > Please check the Web UI for statistics about the checkpoints: > > https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/checkpoint_monitoring.html > > > You're going through a lot of messages in between the checkpoints. Which > state backend do you use? Please try re-running your job with the file > system state backend (FsStateBackend) or the RocksDB state backend > (RocksDBStateBackend). For the RocksDB state backend you will have to > add the RocksDB dependency. The file system backend should work out of > the box, just specify a path and set > FlinkPipelineOptions#setStateBackend(..). See: > > https://ci.apache.org/projects/flink/flink-docs-release-1.5/ops/state/state_backends.html > > Next, I could supply you with a custom Beam version which logs more > debug information. > > Best, > Max > > On 13.09.18 16:40, Encho Mishinev wrote: > > Hello Max, > > > > I am currently performing more tests on it and will follow-up with > > anything I find. > > > > Currently I have the following observations: > > > > Whenever there are few (relative to the parallelism) messages left in a > > pubsub topic the checkpointing length becomes very long. I have tried > > this with different parallelism. My usual set for testing is 13 task > > managers with 4 task slots eac, 52 parallelism for the job and > > checkpointing every 60s. I've done three runs on a subscription filled > > with about 122,000,000 messages. The job works fast going through about > > 1,500,000 messages/minute until it reaches about 120,000,000 or so, when > > it progressively slows down. Checkpointing length increases from an > > average of 50-60s to 2:30min-3min. When about a few hundred thousand > > messages are left the job mostly does long checkpoints and no work. > > Messages pass through but seemingly forever. > > > > The first job I stopped right when it processed more messages than I had > > loaded. The subscription afterwards had 52 000 unacknowledged messages. > > > > Another job with the same approach had 87 000 unacknowledged messages. > > > > A third job I left over 30 minutes after it had processed more messages > > than I had loaded. It worked very slowly with long checkpoints and > > processed a few hundred thousand messages in total over the 30 minute > > period. That subscription then had only 235 unacknowledged messages left. > > > > I have put large acknowledgement deadline for the subscriptions so that > > the checkpointing time is less than the deadline (otherwise the messages > > are naturally resent and can't be acknowledged), that unfortunately is > > not the problem. > > > > I then tried running the whole thing with parallelism of 1 and about 100 > > 000 messages. The job started fast once again, doing a few thousand a > > second and doing all checkpoints in under 3s. Upon reaching about 90 000 > > it again started to slow down. This time it slowly reached it's goal and > > there were actually no unacknowledged messages, but the last 10 000 > > messages were processed dreadfully slowly and one checkpoint during that > > period took 45s (compared to tens of checkpoints under 3s before that). > > > > I am not sure how to check how many messages get acknowledged per > > checkpoint. > > I'm open to trying new runs and sharing the results - let me know if you > > want me to try and run the job with some specific parameters. > > > > Thanks for the help, > > Encho > > > > On Thu, Sep 13, 2018 at 5:20 PM Maximilian Michels <[email protected] > > <mailto:[email protected]>> wrote: > > > > That is indeed strange. Would you be able to provide some debugging > > information, e.g. how many message get acked for each checkpoint? > > > > What is the parallelism of your job? > > > > Thanks, > > Max > > > > On 12.09.18 12:57, Encho Mishinev wrote: > > > Hello Max, > > > > > > Thanks for the answer. My guess was that they are acknowledged at > > > completion of Flink's checkpoints, but wanted to make sure since > > that > > > doesn't explain my problem. > > > > > > Whenever a subscription is nearly empty the job gets slower > > overall and > > > the Flink's checkpoints start taking much more time (thrice or > more) > > > even though their state is much smaller, and of course, there > always > > > seem to be messages cycling over and over again. > > > > > > If you have any clue at all why this might be, let me know. > > > > > > Thanks for the help, > > > Encho > > > > > > On Tue, Sep 11, 2018 at 1:45 PM Maximilian Michels > > <[email protected] <mailto:[email protected]> > > > <mailto:[email protected] <mailto:[email protected]>>> wrote: > > > > > > Hey Encho, > > > > > > The Flink Runner acknowledges messages through PubSubIO's > > > `CheckpointMark#finalizeCheckpoint()` method. > > > > > > The Flink Runner wraps the PubSubIO source via the > > > UnboundedSourceWrapper. When Flink takes a checkpoint of the > > running > > > Beam streaming job, the wrapper will retrieve the > > CheckpointMarks from > > > the PubSubIO source. > > > > > > When the Checkpoint is completed, there is a callback which > > informs the > > > wrapper (`notifyCheckpointComplete()`) and calls > > `finalizeCheckpoint()` > > > on all the generated CheckpointMarks. > > > > > > Hope that helps debugging your problem. I don't have an > > explanation why > > > this doesn't work for the last records in your PubSub queue. > It > > > shouldn't make a difference for how the Flink Runner does > > checkpointing. > > > > > > Best, > > > Max > > > > > > On 10.09.18 18:17, Encho Mishinev wrote: > > > > Hello, > > > > > > > > I am using Flink runner with Apache Beam 2.6.0. I was > > wondering > > > if there > > > > is information on when exactly the runner acknowledges a > > pubsub > > > message > > > > when reading from PubsubIO? > > > > > > > > My problem is that whenever there are a few messages left > in a > > > > subscription my streaming job never really seems to > > acknowledge them > > > > all. For example is a subscription has 100,000,000 > messages in > > > total, > > > > the job will go through about 99,990,000 and then keep > reading > > > the last > > > > few thousand and seemingly never acknowledge them. > > > > > > > > Some clarity on when the acknowledgement happens in the > > pipeline > > > might > > > > help me debug this problem. > > > > > > > > Thanks! > > > > > >
