Hi all, we have multiple Flink session clusters, and I was wondering why a Flink job would lose its state when the job gets redeployed? i.e., say there was a job by name "flink-job", if I cancel this job and re-submit it, the new job seems to be starting from fresh. Instead, should it not start from its previous checkpoint? I don't see any checkpoints restored. Can someone help me understand this?

When a Flink job is cancelled and then re-submitted, by default, it will indeed start from scratch unless explicitly configured to restore from a previous savepoint or checkpoint. The reason why is because checkpoints and savepoints are mechanisms in Flink for fault tolerance, but they are used differently during job redeployment. Checkpoints are periodic and automatic, designed primarily for recovery in case of failures. When you cancel a job and restart it, Flink does not automatically restore from the last checkpoint by default. Checkpoints are meant for automatic recovery within the lifecycle of a running job.

When you cancel a job and restart it, Flink does not automatically restore from the last checkpoint by default.

Thanks Draco, what do I need to do to have the job restore from a checkpoint on resubmit?

Concerning savepoints on a session cluster, yes, it is absolutely possible and common practice to use savepoints with both session and per-job clusters in Flink. Savepoints are a versatile feature designed for maintaining the state of applications across upgrades, configuration changes, or even switching between session and per-job clusters. They are not limited to a specific type of cluster deployment.

OK, could you please point me to some documentation on how I can use checkpoints and savepoints to restore state on job resubmit?

To clarify, if you cancel a Flink job and then want to resume from where it left off, you must use savepoints, not regular checkpoints, due to the nature of their intended use cases—checkpoints for automatic recovery within a continuous job run, and savepoints for planned maintenance or state migrations.

But restoring from savepoints causes a lot of duplicate data to be reprocessed, right? As in we usually take a savepoint every hour or so. By restoring from a savepoint, it'll be as though we went bck an hour to process data. We take checkpoints every 30s, this would have been much better if we were to restart from a checkpoint

How are you cancelling?

Using flink cancel API

I think frequent checkpointing is a good idea actual failures errors etc. Unplanned interruptions. But for a planned cancellations savepoints seem to be what’s recommended by the documentation

Thank you!

Good luck on it!

One last question: if a job restarts by itself, say due to some EKS pod shuffling that caused a task manager to be taken out and reallocated elsewhere, I'm assuming the job would restart and restore from a previous checkpoint, right?

Yes, you are correct. If a Flink job is set up with checkpointing enabled and it experiences a failure or restart due to events like an Elastic Kubernetes Service (EKS) pod rescheduling, Flink is designed to automatically restore from the most recent successful checkpoint. This is a core aspect of Flink’s fault-tolerance mechanism.

great, thank you