Posting this here from other channel https://apache-flink.slack.com/archives/C03JKTFFX0S/p1685729088580199

It would help to know if you are doing joins that depend on event-time semantics — e.g., temporal joins, or interval joins. And it would be interesting to how checkpointing is configured. There are many things that might be helpful. You might want to read up on • watermark alignment (https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#watermark-alignment-_beta_) • unaligned checkpoints (https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/checkpointing_under_backpressure/#unaligned-checkpoints) • the changelog state backend (https://flink.apache.org/2022/05/30/improving-speed-and-stability-of-checkpointing-with-generic-log-based-incremental-checkpoints/)

Thanks @David Anderson for your quick reply. We are using temporal joins, Here is our checkpoint configuration, I will take a look at these links above