Today, we take a checkpoint periodically based on some factors, the main ones being "have we accumulated enough state size" with some additional bounds like atleast once every hour.

For the non-window case where the state size is typically less, we end up taking a lot of checkpoints, something like once every 5-6 minutes.

We can will change this to also use processing time (i.e. "cpu time") spent in the pipeline as an additional signal. So the form of the heuristic will be "atleast x secs or atleast y bytes (of state size)" with the additional caps like atleast once per hour and so on.