Hey everyone, We use flink runner (v1.13) with beam pipelines (with kafka topic as input source) and use Flink kubernetes operator for deployment . We seem to have encountered a lost message due to a message processing failure causing a restart in the pipeline. We have enabled flink checkpoints with S3 storage, we have also set the beam setting to commit kafka offset on finalize . We enabled unaligned checkpointing(few weeks ago) after we saw long delays in checkpointing. Yesterday we saw an exception in the pipeline, which caused it to restart. And flink restored the state from the last checkpoint but to our surprise , the failed message was not retried again but the offset had moved to the next position. Our expectation was that the offset would not move until the message has been processed. This is the first time we are seeing this issue, where we had lost a message because the offsets moved. When can this type of data loss happen ? Is it related to Unaligned checkpointing? Any help would be greatly appreciated. cc @Arun