[FLINK-4717] Add CancelJobWithSavepoint

[uce]

This builds on top of #2608.

• Adds CancelJobWithSavepoint message, which triggers a savepoint before cancelling the respective job. The periodic checkpoint scheduler is stopped before doing this so that no checkpoints happen between the savepoint and cancellation. Note that users could manually trigger another savepoint while cancel-with-savepoint happens, but I think this is tolerable.
• Adds -s [targetDirectory] option to CLI cancel command:
  • Regular cancelling:bin/flink cancel <jobID>
  • Cancel with savepoint to default dir: bin/flink cancel -s <jobID>
  • Cancel with savepoint to target dir: bin/flink cancel -s [targetDir] <jobID>

[uce]

To help review, the main change is in JobManager and CliFrontend. Some questions there may be:

• Should we keep the behaviour that the job is not cancelled if the savepoint fails? My reasoning for this was that the user might otherwise accidentally loose state when cancelling the job.
• Does the way that the cancellation is specified on the CLI make sense? Should we keep the -s option or add another proper command like bin/flink cancel-with-savepoint?

Furthermore, only the 2nd commit is relevant in this PR as the 1st one is from #2608.

[tillrohrmann]

Changes look again very good @uce :-)

I had a comment concerning stopping the checkpoint scheduler which should be fixed imo.

A minor comment concerned the order of cancel -s arguments. But I understand the reason why it is how it is. I prefer to not have a dedicated command for the cancel with savepoint. Thus, I like the approach you've chosen.

Concerning not executing the cancel operation if the savepoint fails, I think that this is the right way to go (with the same arguments as you've presented).

Apart from that +1 for merging.

[tillrohrmann]

It is a little bit confusing that the savepoint command has a different argument order.

[uce]

We could also change the savepoint trigger command to savepoint -t [targetDir] <jobID>.

Or we could allow cancel <jobID> [targetDirectory] in addition to the -s option variant, but that might be slightly confusing, too. We need the -s option without an argument for cancel with savepoint to the default target.

What do you think?

[tillrohrmann]

I think it's not enough to simply call stopCheckpointScheduler. If I'm not mistaken, then the following could happen: You call stopCheckpointScheduler which will try to cancel the last currentPeriodicTrigger. Now assume that the last TimerTask to trigger the next checkpoint has just been triggered but not executed (just before cancelling it). Now the stopCheckpointScheduler finishes without the TimerTask having completed. Now the TimerTask can still trigger a checkpoint even though we've stopped the checkpoint scheduler.

The way to fix this (admittedly academic corner case), is to filter out outdated TimerTask calls in the CheckpointCoordinator by having a kind of fencing tokens for the trigger checkpoint calls.

[uce]

Thanks for this review, too. Good catch with the broken stopCheckpointScheduler. I will add your proposed fix as a separate commit.

[uce]

Addressed the issue and found another bug in #2608 that I've fixed in f769e8e. If Travis gives the green light, I will rebase on #2608 and merge this later today.

[uce]

Local Travis build passed (https://travis-ci.org/uce/flink/builds/167108720). I'm going to rebase and merge this later.