I am running a Flink job that reads data from Kafka, processes it into a Flink Row object, and writes it to an Iceberg Sink. To deploy new code changes, I restart the job from the latest savepoint, which is committed at the Kafka source. This savepoint stores the most recent Kafka offset read by the job, allowing the subsequent run to continue from the previous offset, ensuring transparent failure recovery.
Recently, I encountered a scenario where data was lost in the final output after a job restart. Here’s what happened:

1. The job reads from Kafka starting at offset 1 and writes the data to an Iceberg table backed by S3.
2. It then reads from offset 2, but the writes to the Iceberg table are delayed due to backpressure or network issues.
3. The job proceeds to read from offset 3.
4. Before the data from offset 2 is written to S3, I restart the job.
   After the restart, the job begins reading from offset 3, resulting in the loss of data from offset 2, which was never written to S3.

Is there a workaround for this problem, or is it an inherent limitation of the Iceberg Sink?