Repository: apache/iceberg (language: python)

Issue:
Flink: Fix duplicate data with upsert writer in case of aborted checkpoints
resolve #10431 


Discussion / hints:
@zhongqishang: Why would this solve the issue?

In my mental model, this only changes when the files are closed. The files are added in the `IcebergFilesCommitter` operator state in the `snapshotState` and to the table in the `notifyCheckpointComplete`, so I think the issue could be about how calling these methods are orchestrated.
> @zhongqishang: Why would this solve the issue?

I am trying to solve this problem. This scenario is hard to reproduce.

I want to re-express the background of the problem. The checkpoint was triggered twice in succession, and the previous checkpoint was canceled. Cancelling the checkpoint did not cause the flink job to fail. Therefore, the results of prepareSnapshotPreBarrier twice were generated in one commit.

Moving to snapshot is just to ensure that flush success is bound to checkpoint success, so that the above scenario will not occur. 

Of course I'm not sure if this will solve the problem or if it will cause other problems. 

@pvary Thank you very much for your guidance.
@zhongqishang: We have 2 data files with the same data. I suspect that the 1st data file is generated in the 1st checkpoint, and the 2nd data file is generated in the 2nd checkpoint. Could you confirm that the writers are restarted, or the source is restarted? Is the source is stateful, and the state of the source is handled correctly?

Also, could you please check if the `IcebergFilesCommitter` operator is restarted, or not in the meantime?
> @zhongqishang: We have 2 data files with the same data. I suspect that the 1st data file is generated in the 1st checkpoint, and the 2nd data file is generated in the 2nd checkpoint. Could you confirm that the writers are restarted, or the source is restarted? Is the source is stateful, and the state of the source is handled correctly?

Just like you said.

> Also, could you please check if the `IcebergFilesCommitter` operator is restarted, or not in the meantime?

Operator not restarted. The flink job is normal.
> > @zhongqishang: We have 2 data files with the same data. I suspect that the 1st data file is generated in the 1st checkpoint, and the 2nd data file is generated in the 2nd checkpoint. Could you confirm that the writers are restarted, or the source is restarted? Is the source is stateful, and the state of the source is handled correctly?
> 
> Just like you said.

Sadly, I don't understand your answer fully.
What do we know about the writers/sources? Are they restarted?

> > Also, could you please check if the `IcebergFilesCommitter` operator is restarted, or not in the meantime?
> 
> Operator not restarted. The flink job is normal.

Do we have logs like these?
https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L204C10-L204C63
https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L240
https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L240
> > > @zhongqishang: We have 2 data files with the same data. I suspect that the 1st data file is generated in the 1st checkpoint, and the 2nd data file is generated in the 2nd checkpoint. Could you confirm that the writers are restarted, or the source is restarted? Is the source is stateful, and the state of the source is handled correctly?
> > 
> > 
> > Just like you said.
> 
> Sadly, I don't understand your answer fully. What do we know about the writers/sources? Are they restarted?

I'm sorry for not expressing myself clearly. I confirmed that the 1st data file is generated in the 1st checkpoint, and the 2nd data file is generated in the 2nd checkpoint. I confirmed that the writers and the source are not restarted.

> Do we have logs like these? https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L204C10-L204C63 https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L240 https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L240

Not found the logs. 

I think the checkpoint is not executing `snapshotState` and the checkpoint is aborted after running prepareSnapshotPreBarrier.
Thanks for the clarification around the restarts.

> Not found the logs.
> 
> I think the checkpoint is not executing `snapshotState` and the checkpoint is aborted after running prepareSnapshotPreBarrier.

That's strange. Could you please double check (log levels,  TM logs)? These are the logs which should be there if the changes are committed by the Flink  job.

If the logs aren't there, you can cross-check by the snapshot summary. Here please check if the flink jobId, checkpointId, operatorId is there or missing on the commit.
> That's strange. Could you please double check (log levels, TM logs)? These are the logs which should be there if the changes are committed by the Flink job.

> https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L204C10-L204C63 [https://github.com/apache/iceberg/blob/apache-iceberg-](https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L240)

For https://github.com/apache/iceberg/issues/10431#issuecomment-2160846434

log level is `INFO`, A normal checkpoint has a commit log. 
```
Start to flush snapshot state to state backend, table: iceberg_catalog.ods_iceberg_db.xx, checkpointId: 86446
// The log of aborted checkpoint 86447 is missing
Start to flush snapshot state to state backend, table: iceberg_catalog.ods_iceberg_db.xx, checkpointId: 86448
```


> https://github.com/apache/iceberg/blob/apache-iceberg-1.2.1/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L240

This log does not exist in the all tm log.



> log level is `INFO`, A normal checkpoint has a commit log.
> 
> ```
> Start to flush snapshot state to state backend, table: iceberg_catalog.ods_iceberg_db.xx, checkpointId: 86446
> // The log of aborted checkpoint 86447 is missing
> Start to flush snapshot state to state backend, table: iceberg_catalog.ods_iceberg_db.xx, checkpointId: 86448
> ```

This means, that the `IcebergFilesCommitter` checkpointing is not started at all.
Since the files are created we could conclude that the `IcebergStreamWriter.flush` is called. Also, since the next write didn't fail, I would expect that the `prepareSnapshotPreBarrier` is finished correctly. That would mean, that the created data files are collected in the `flush` method... So everything seems normal.

Again.. are we sure that we don't get duplicated records from the input stream for whatever reasons?


> Again.. are we sure that we don't get duplicated records from the input stream for whatever reasons?

The task has upsert enabled. 
All data will be entered multiple times from the input stream. I have hundreds of tasks. Only in the scenario I described will two pieces of data with the same ID be displayed in the query results. Upsert should only keep one.

> We have 2 data files with the same data. I suspect that the 1st data file is generated in the 1st checkpoint, and the 2nd data file is generated in the 2nd checkpoint. 

As you described before. 

The problem occurs when the result of 2 `prepareSnapshotPreBarriers` appears in one commit.

> https://github.com/apache/iceberg/issues/10431#issuecomment-2148804889

As I described before, in the one commit, it will query 2 pieces of data with the same id.
So in summary:
- The duplicated data is expected as this is coming from the input stream
- The issue is that the `WriteResult` does not contain the checkpointId, and the IcebergFilesCommitter commits the changes for 2 checkpoints in a single snapshot.

So the fix should be something along these lines:
- Add checkpointId to the `WriteResult`
- Change the `IcebergFileCommitter` to consider the checkpointId in the `WriteResult`s and generate multiple commits if needed.

@rodmeneses: Is this issue solved naturally by the new SinkV2 implemetation?
@pvary 
Adding checkpointId to `WriteResult` will cause changes to the Public API and is incompatible with serialization. 
So I added a wrapper class `FlinkWriteResult` to Flink to store checkpointId. 
WDYT? 

Can you give me some advice?

Let me check what compatibility issues we are creating by changing the `WriteResult` to `FlinkWriteResult`.
Otherwise looks good to me
@stevenzwu:
How should we handle this situation?
`WriteResult` is used between the communication of the writer and the committer. `WriteResult` is part of the Iceberg API, so we can not change it.

Without changing the `WriteResult` we can't fix the issue (at least, I can't see a way right now).

If we change the type we use between the communication of the writer and the committer, then the users who are using unaligned checkpoints in their job with FlinkSink, need to restart the job without state. For users who are using aligned checkpoints this is a compatible fix.

Do you have better idea, how to fix this? If not, how do we communicate this? (The job restart will fail, but it would be nicer if we document this somewhere)
I'm still thinking about the best way moving forward. Maybe we should ask the dev lists opinion about the compatibility issue. Let's see how widely the unaligned checkpoints are used in the community. 
> I'm still thinking about the best way moving forward. Maybe we should ask the dev lists opinion about the compatibility issue. Let's see how widely the unaligned checkpoints are used in the community.

Thanks @pvary the continued follow-up. It's a very good idea that ask the dev lists the unaligned checkpoints are used in the community.

My English output is not accurate, can you send an email about this?
@zhongqishang: Finally found time to start the upstream discussion: https://lists.apache.org/thread/n5c85hd7psf2tmgych6scmynonscp2q4
@zhongqishang @pvary I have a uber question.

let's say checkpoint N was cancelled or timed out and checkpoint N+1 completed successfully. In this case, do we know all the writer subtasks have flushed data files for checkpoint N and all write results have all been received by the committer?
> @zhongqishang @pvary I have a uber question.
> 
> let's say checkpoint N was cancelled or timed out and checkpoint N+1 completed successfully. In this case, do we know all the writer subtasks have flushed data files for checkpoint N and all write results have all been received by the committer?

I think we only need to be sure, that every writer which has successfully closed on a given checkpoint is added in an Iceberg commit for the given checkpoint.
If some of the writers are not closed, then they will keep collecting consistent data, and the results of the next checkpoint will be consistent.

What we should be aware, is that the Iceberg commit for the non-successful checkpoint might not contain everything which is received up to the checkpoint time. I think this is ok, since the users could check if the given checkpoint was successful or not
@zhongqishang: Thanks for the patience. I revisited the PR again, and found one potential issue. Could you please check it?
I think we are getting there. Thanks @gyfora for the clarification.
@zhongqishang: Could you please rebase?
> @zhongqishang: Could you please rebase?

@pvary Already rebased, PTAL.
thanks @zhongqishang for the root cause and fix. and thanks @pvary for the review.

@zhongqishang please create a back port PR to 1.18 and 1.20. 

also let's create a standing PR for the next 1.7 release notes (`site/docs/releases.md`) to clarify the state incompatibility and savepoint is required. we can make that PR for the 1.7 milestone and merge it when 1.7 is ready. 

Fix patch:
diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index 2256d1e874ce..5cd43a46de37 100644
--- a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++ b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -67,7 +67,6 @@
 import org.apache.iceberg.flink.sink.shuffle.StatisticsOrRecord;
 import org.apache.iceberg.flink.sink.shuffle.StatisticsType;
 import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
-import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -420,7 +419,7 @@ private <T> DataStreamSink<T> chainIcebergOperators() {
           distributeDataStream(rowDataInput, equalityFieldIds, flinkRowType, writerParallelism);
 
       // Add parallel writers that append rows to files
-      SingleOutputStreamOperator<WriteResult> writerStream =
+      SingleOutputStreamOperator<FlinkWriteResult> writerStream =
           appendWriter(distributeStream, flinkRowType, equalityFieldIds, writerParallelism);
 
       // Add single-parallelism committer that commits files
@@ -487,7 +486,7 @@ private <T> DataStreamSink<T> appendDummySink(
     }
 
     private SingleOutputStreamOperator<Void> appendCommitter(
-        SingleOutputStreamOperator<WriteResult> writerStream) {
+        SingleOutputStreamOperator<FlinkWriteResult> writerStream) {
       IcebergFilesCommitter filesCommitter =
           new IcebergFilesCommitter(
               tableLoader,
@@ -507,7 +506,7 @@ private SingleOutputStreamOperator<Void> appendCommitter(
       return committerStream;
     }
 
-    private SingleOutputStreamOperator<WriteResult> appendWriter(
+    private SingleOutputStreamOperator<FlinkWriteResult> appendWriter(
         DataStream<RowData> input,
         RowType flinkRowType,
         List<Integer> equalityFieldIds,
@@ -545,11 +544,11 @@ private SingleOutputStreamOperator<WriteResult> appendWriter(
       IcebergStreamWriter<RowData> streamWriter =
           createStreamWriter(tableSupplier, flinkWriteConf, flinkRowType, equalityFieldIds);
 
-      SingleOutputStreamOperator<WriteResult> writerStream =
+      SingleOutputStreamOperator<FlinkWriteResult> writerStream =
           input
               .transform(
                   operatorName(ICEBERG_STREAM_WRITER_NAME),
-                  TypeInformation.of(WriteResult.class),
+                  TypeInformation.of(FlinkWriteResult.class),
                   streamWriter)
               .setParallelism(writerParallelism);
       if (uidPrefix != null) {
diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
new file mode 100644
index 000000000000..317fb169ae1b
--- /dev/null
+++ b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import java.io.Serializable;
+import org.apache.iceberg.io.WriteResult;
+
+public class FlinkWriteResult implements Serializable {
+  private final long checkpointId;
+  private final WriteResult writeResult;
+
+  public FlinkWriteResult(long checkpointId, WriteResult writeResult) {
+    this.checkpointId = checkpointId;
+    this.writeResult = writeResult;
+  }
+
+  public long checkpointId() {
+    return checkpointId;
+  }
+
+  public WriteResult writeResult() {
+    return writeResult;
+  }
+}
diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
index b9bceaa9311d..7108c2008341 100644
--- a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
+++ b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
@@ -63,7 +63,7 @@
 import org.slf4j.LoggerFactory;
 
 class IcebergFilesCommitter extends AbstractStreamOperator<Void>
-    implements OneInputStreamOperator<WriteResult, Void>, BoundedOneInput {
+    implements OneInputStreamOperator<FlinkWriteResult, Void>, BoundedOneInput {
 
   private static final long serialVersionUID = 1L;
   private static final long INITIAL_CHECKPOINT_ID = -1L;
@@ -96,7 +96,7 @@ class IcebergFilesCommitter extends AbstractStreamOperator<Void>
 
   // The completed files cache for current checkpoint. Once the snapshot barrier received, it will
   // be flushed to the 'dataFilesPerCheckpoint'.
-  private final List<WriteResult> writeResultsOfCurrentCkpt = Lists.newArrayList();
+  private final Map<Long, List<WriteResult>> writeResultsSinceLastSnapshot = Maps.newHashMap();
   private final String branch;
 
   // It will have an unique identifier for one job.
@@ -212,7 +212,8 @@ public void snapshotState(StateSnapshotContext context) throws Exception {
 
     // Update the checkpoint state.
     long startNano = System.nanoTime();
-    dataFilesPerCheckpoint.put(checkpointId, writeToManifest(checkpointId));
+    writeToManifestUptoLatestCheckpoint(checkpointId);
+
     // Reset the snapshot state to the latest state.
     checkpointsState.clear();
     checkpointsState.add(dataFilesPerCheckpoint);
@@ -220,8 +221,6 @@ public void snapshotState(StateSnapshotContext context) throws Exception {
     jobIdState.clear();
     jobIdState.add(flinkJobId);
 
-    // Clear the local buffer for current checkpoint.
-    writeResultsOfCurrentCkpt.clear();
     committerMetrics.checkpointDuration(
         TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano));
   }
@@ -426,30 +425,45 @@ private void commitOperation(
   }
 
   @Override
-  public void processElement(StreamRecord<WriteResult> element) {
-    this.writeResultsOfCurrentCkpt.add(element.getValue());
+  public void processElement(StreamRecord<FlinkWriteResult> element) {
+    FlinkWriteResult flinkWriteResult = element.getValue();
+    List<WriteResult> writeResults =
+        writeResultsSinceLastSnapshot.computeIfAbsent(
+            flinkWriteResult.checkpointId(), k -> Lists.newArrayList());
+    writeResults.add(flinkWriteResult.writeResult());
   }
 
   @Override
   public void endInput() throws IOException {
     // Flush the buffered data files into 'dataFilesPerCheckpoint' firstly.
-    long currentCheckpointId = Long.MAX_VALUE;
-    dataFilesPerCheckpoint.put(currentCheckpointId, writeToManifest(currentCheckpointId));
-    writeResultsOfCurrentCkpt.clear();
-
+    long currentCheckpointId = IcebergStreamWriter.END_INPUT_CHECKPOINT_ID;
+    writeToManifestUptoLatestCheckpoint(currentCheckpointId);
     commitUpToCheckpoint(dataFilesPerCheckpoint, flinkJobId, operatorUniqueId, currentCheckpointId);
   }
 
+  private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
+    if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+      dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
+    }
+
+    for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
+        writeResultsSinceLastSnapshot.entrySet()) {
+      dataFilesPerCheckpoint.put(
+          writeResultsOfCheckpoint.getKey(),
+          writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+    }
+
+    // Clear the local buffer for current checkpoint.
+    writeResultsSinceLastSnapshot.clear();
+  }
+
   /**
    * Write all the complete data files to a newly created manifest file and return the manifest's
    * avro serialized bytes.
    */
-  private byte[] writeToManifest(long checkpointId) throws IOException {
-    if (writeResultsOfCurrentCkpt.isEmpty()) {
-      return EMPTY_MANIFEST_DATA;
-    }
-
-    WriteResult result = WriteResult.builder().addAll(writeResultsOfCurrentCkpt).build();
+  private byte[] writeToManifest(long checkpointId, List<WriteResult> writeResults)
+      throws IOException {
+    WriteResult result = WriteResult.builder().addAll(writeResults).build();
     DeltaManifests deltaManifests =
         FlinkManifestUtil.writeCompletedFiles(
             result, () -> manifestOutputFileFactory.create(checkpointId), spec);
diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
index 9ea0349fb057..bb5efe982ee1 100644
--- a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
+++ b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
@@ -29,10 +29,11 @@
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 
-class IcebergStreamWriter<T> extends AbstractStreamOperator<WriteResult>
-    implements OneInputStreamOperator<T, WriteResult>, BoundedOneInput {
+class IcebergStreamWriter<T> extends AbstractStreamOperator<FlinkWriteResult>
+    implements OneInputStreamOperator<T, FlinkWriteResult>, BoundedOneInput {
 
   private static final long serialVersionUID = 1L;
+  static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
 
   private final String fullTableName;
   private final TaskWriterFactory<T> taskWriterFactory;
@@ -63,7 +64,7 @@ public void open() {
 
   @Override
   public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
-    flush();
+    flush(checkpointId);
     this.writer = taskWriterFactory.create();
   }
 
@@ -89,7 +90,7 @@ public void endInput() throws IOException {
     // Note that if the task is not closed after calling endInput, checkpoint may be triggered again
     // causing files to be sent repeatedly, the writer is marked as null after the last file is sent
     // to guard against duplicated writes.
-    flush();
+    flush(END_INPUT_CHECKPOINT_ID);
   }
 
   @Override
@@ -102,7 +103,7 @@ public String toString() {
   }
 
   /** close all open files and emit files to downstream committer operator */
-  private void flush() throws IOException {
+  private void flush(long checkpointId) throws IOException {
     if (writer == null) {
       return;
     }
@@ -110,7 +111,7 @@ private void flush() throws IOException {
     long startNano = System.nanoTime();
     WriteResult result = writer.complete();
     writerMetrics.updateFlushResult(result);
-    output.collect(new StreamRecord<>(result));
+    output.collect(new StreamRecord<>(new FlinkWriteResult(checkpointId, result)));
     writerMetrics.flushDuration(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano));
 
     // Set writer to null to prevent duplicate flushes in the corner case of
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index be2a8db03097..c53431490984 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -67,7 +67,6 @@
 import org.apache.iceberg.flink.sink.shuffle.StatisticsOrRecord;
 import org.apache.iceberg.flink.sink.shuffle.StatisticsType;
 import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
-import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -421,7 +420,7 @@ private <T> DataStreamSink<T> chainIcebergOperators() {
           distributeDataStream(rowDataInput, equalityFieldIds, flinkRowType, writerParallelism);
 
       // Add parallel writers that append rows to files
-      SingleOutputStreamOperator<WriteResult> writerStream =
+      SingleOutputStreamOperator<FlinkWriteResult> writerStream =
           appendWriter(distributeStream, flinkRowType, equalityFieldIds, writerParallelism);
 
       // Add single-parallelism committer that commits files
@@ -488,7 +487,7 @@ private <T> DataStreamSink<T> appendDummySink(
     }
 
     private SingleOutputStreamOperator<Void> appendCommitter(
-        SingleOutputStreamOperator<WriteResult> writerStream) {
+        SingleOutputStreamOperator<FlinkWriteResult> writerStream) {
       IcebergFilesCommitter filesCommitter =
           new IcebergFilesCommitter(
               tableLoader,
@@ -508,7 +507,7 @@ private SingleOutputStreamOperator<Void> appendCommitter(
       return committerStream;
     }
 
-    private SingleOutputStreamOperator<WriteResult> appendWriter(
+    private SingleOutputStreamOperator<FlinkWriteResult> appendWriter(
         DataStream<RowData> input,
         RowType flinkRowType,
         List<Integer> equalityFieldIds,
@@ -546,11 +545,11 @@ private SingleOutputStreamOperator<WriteResult> appendWriter(
       IcebergStreamWriter<RowData> streamWriter =
           createStreamWriter(tableSupplier, flinkWriteConf, flinkRowType, equalityFieldIds);
 
-      SingleOutputStreamOperator<WriteResult> writerStream =
+      SingleOutputStreamOperator<FlinkWriteResult> writerStream =
           input
               .transform(
                   operatorName(ICEBERG_STREAM_WRITER_NAME),
-                  TypeInformation.of(WriteResult.class),
+                  TypeInformation.of(FlinkWriteResult.class),
                   streamWriter)
               .setParallelism(writerParallelism);
       if (uidPrefix != null) {
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
new file mode 100644
index 000000000000..317fb169ae1b
--- /dev/null
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import java.io.Serializable;
+import org.apache.iceberg.io.WriteResult;
+
+public class FlinkWriteResult implements Serializable {
+  private final long checkpointId;
+  private final WriteResult writeResult;
+
+  public FlinkWriteResult(long checkpointId, WriteResult writeResult) {
+    this.checkpointId = checkpointId;
+    this.writeResult = writeResult;
+  }
+
+  public long checkpointId() {
+    return checkpointId;
+  }
+
+  public WriteResult writeResult() {
+    return writeResult;
+  }
+}
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
index 622daa808897..609deb621f43 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
@@ -61,7 +61,7 @@
 import org.slf4j.LoggerFactory;
 
 class IcebergFilesCommitter extends AbstractStreamOperator<Void>
-    implements OneInputStreamOperator<WriteResult, Void>, BoundedOneInput {
+    implements OneInputStreamOperator<FlinkWriteResult, Void>, BoundedOneInput {
 
   private static final long serialVersionUID = 1L;
   private static final long INITIAL_CHECKPOINT_ID = -1L;
@@ -94,7 +94,7 @@ class IcebergFilesCommitter extends AbstractStreamOperator<Void>
 
   // The completed files cache for current checkpoint. Once the snapshot barrier received, it will
   // be flushed to the 'dataFilesPerCheckpoint'.
-  private final List<WriteResult> writeResultsOfCurrentCkpt = Lists.newArrayList();
+  private final Map<Long, List<WriteResult>> writeResultsSinceLastSnapshot = Maps.newHashMap();
   private final String branch;
 
   // It will have an unique identifier for one job.
@@ -210,7 +210,8 @@ public void snapshotState(StateSnapshotContext context) throws Exception {
 
     // Update the checkpoint state.
     long startNano = System.nanoTime();
-    dataFilesPerCheckpoint.put(checkpointId, writeToManifest(checkpointId));
+    writeToManifestUptoLatestCheckpoint(checkpointId);
+
     // Reset the snapshot state to the latest state.
     checkpointsState.clear();
     checkpointsState.add(dataFilesPerCheckpoint);
@@ -218,8 +219,6 @@ public void snapshotState(StateSnapshotContext context) throws Exception {
     jobIdState.clear();
     jobIdState.add(flinkJobId);
 
-    // Clear the local buffer for current checkpoint.
-    writeResultsOfCurrentCkpt.clear();
     committerMetrics.checkpointDuration(
         TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano));
   }
@@ -403,30 +402,45 @@ private void commitOperation(
   }
 
   @Override
-  public void processElement(StreamRecord<WriteResult> element) {
-    this.writeResultsOfCurrentCkpt.add(element.getValue());
+  public void processElement(StreamRecord<FlinkWriteResult> element) {
+    FlinkWriteResult flinkWriteResult = element.getValue();
+    List<WriteResult> writeResults =
+        writeResultsSinceLastSnapshot.computeIfAbsent(
+            flinkWriteResult.checkpointId(), k -> Lists.newArrayList());
+    writeResults.add(flinkWriteResult.writeResult());
   }
 
   @Override
   public void endInput() throws IOException {
     // Flush the buffered data files into 'dataFilesPerCheckpoint' firstly.
-    long currentCheckpointId = Long.MAX_VALUE;
-    dataFilesPerCheckpoint.put(currentCheckpointId, writeToManifest(currentCheckpointId));
-    writeResultsOfCurrentCkpt.clear();
-
+    long currentCheckpointId = IcebergStreamWriter.END_INPUT_CHECKPOINT_ID;
+    writeToManifestUptoLatestCheckpoint(currentCheckpointId);
     commitUpToCheckpoint(dataFilesPerCheckpoint, flinkJobId, operatorUniqueId, currentCheckpointId);
   }
 
+  private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
+    if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+      dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
+    }
+
+    for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
+        writeResultsSinceLastSnapshot.entrySet()) {
+      dataFilesPerCheckpoint.put(
+          writeResultsOfCheckpoint.getKey(),
+          writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+    }
+
+    // Clear the local buffer for current checkpoint.
+    writeResultsSinceLastSnapshot.clear();
+  }
+
   /**
    * Write all the complete data files to a newly created manifest file and return the manifest's
    * avro serialized bytes.
    */
-  private byte[] writeToManifest(long checkpointId) throws IOException {
-    if (writeResultsOfCurrentCkpt.isEmpty()) {
-      return EMPTY_MANIFEST_DATA;
-    }
-
-    WriteResult result = WriteResult.builder().addAll(writeResultsOfCurrentCkpt).build();
+  private byte[] writeToManifest(long checkpointId, List<WriteResult> writeResults)
+      throws IOException {
+    WriteResult result = WriteResult.builder().addAll(writeResults).build();
     DeltaManifests deltaManifests =
         FlinkManifestUtil.writeCompletedFiles(
             result, () -> manifestOutputFileFactory.create(checkpointId), spec);
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
index 7d86baa14fc2..412d6c7081bf 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
@@ -29,10 +29,11 @@
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 
-class IcebergStreamWriter<T> extends AbstractStreamOperator<WriteResult>
-    implements OneInputStreamOperator<T, WriteResult>, BoundedOneInput {
+class IcebergStreamWriter<T> extends AbstractStreamOperator<FlinkWriteResult>
+    implements OneInputStreamOperator<T, FlinkWriteResult>, BoundedOneInput {
 
   private static final long serialVersionUID = 1L;
+  static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
 
   private final String fullTableName;
   private final TaskWriterFactory<T> taskWriterFactory;
@@ -63,7 +64,7 @@ public void open() {
 
   @Override
   public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
-    flush();
+    flush(checkpointId);
     this.writer = taskWriterFactory.create();
   }
 
@@ -89,7 +90,7 @@ public void endInput() throws IOException {
     // Note that if the task is not closed after calling endInput, checkpoint may be triggered again
     // causing files to be sent repeatedly, the writer is marked as null after the last file is sent
     // to guard against duplicated writes.
-    flush();
+    flush(END_INPUT_CHECKPOINT_ID);
   }
 
   @Override
@@ -102,7 +103,7 @@ public String toString() {
   }
 
   /** close all open files and emit files to downstream committer operator */
-  private void flush() throws IOException {
+  private void flush(long checkpointId) throws IOException {
     if (writer == null) {
       return;
     }
@@ -110,7 +111,7 @@ private void flush() throws IOException {
     long startNano = System.nanoTime();
     WriteResult result = writer.complete();
     writerMetrics.updateFlushResult(result);
-    output.collect(new StreamRecord<>(result));
+    output.collect(new StreamRecord<>(new FlinkWriteResult(checkpointId, result)));
     writerMetrics.flushDuration(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano));
 
     // Set writer to null to prevent duplicate flushes in the corner case of


Test patch:
diff --git a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
index 8faae1b05a4e..3299e7a97776 100644
--- a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
+++ b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
@@ -40,7 +40,6 @@
 import org.apache.iceberg.flink.SimpleDataUtil;
 import org.apache.iceberg.io.BaseTaskWriter;
 import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestTemplate;
@@ -210,8 +209,10 @@ public void testCompressionOrc() throws Exception {
         .containsEntry(TableProperties.ORC_COMPRESSION_STRATEGY, "speed");
   }
 
-  private static OneInputStreamOperatorTestHarness<RowData, WriteResult> createIcebergStreamWriter(
-      Table icebergTable, TableSchema flinkSchema, Map<String, String> override) throws Exception {
+  private static OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
+      createIcebergStreamWriter(
+          Table icebergTable, TableSchema flinkSchema, Map<String, String> override)
+          throws Exception {
     RowType flinkRowType = FlinkSink.toFlinkRowType(icebergTable.schema(), flinkSchema);
     FlinkWriteConf flinkWriteConfig =
         new FlinkWriteConf(
@@ -219,7 +220,7 @@ private static OneInputStreamOperatorTestHarness<RowData, WriteResult> createIce
 
     IcebergStreamWriter<RowData> streamWriter =
         FlinkSink.createStreamWriter(() -> icebergTable, flinkWriteConfig, flinkRowType, null);
-    OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+    OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
         new OneInputStreamOperatorTestHarness<>(streamWriter, 1, 1, 0);
 
     harness.setup();
@@ -230,7 +231,7 @@ private static OneInputStreamOperatorTestHarness<RowData, WriteResult> createIce
 
   private static Map<String, String> appenderProperties(
       Table table, TableSchema schema, Map<String, String> override) throws Exception {
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter(table, schema, override)) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
 
diff --git a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
index 948c7b31430c..ac5babe11943 100644
--- a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
+++ b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
@@ -129,7 +129,8 @@ public void testCommitTxnWithoutDataFiles() throws Exception {
     long timestamp = 0;
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -161,7 +162,8 @@ public void testMaxContinuousEmptyCommits() throws Exception {
     JobID jobId = new JobID();
     long checkpointId = 0;
     long timestamp = 0;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
 
@@ -176,8 +178,8 @@ public void testMaxContinuousEmptyCommits() throws Exception {
     }
   }
 
-  private WriteResult of(DataFile dataFile) {
-    return WriteResult.builder().addDataFiles(dataFile).build();
+  private FlinkWriteResult of(long checkpointId, DataFile dataFile) {
+    return new FlinkWriteResult(checkpointId, WriteResult.builder().addDataFiles(dataFile).build());
   }
 
   @TestTemplate
@@ -193,7 +195,8 @@ public void testCommitTxn() throws Exception {
 
     JobID jobID = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobID)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobID)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -204,7 +207,7 @@ public void testCommitTxn() throws Exception {
       for (int i = 1; i <= 3; i++) {
         RowData rowData = SimpleDataUtil.createRowData(i, "hello" + i);
         DataFile dataFile = writeDataFile("data-" + i, ImmutableList.of(rowData));
-        harness.processElement(of(dataFile), ++timestamp);
+        harness.processElement(of(i, dataFile), ++timestamp);
         rows.add(rowData);
 
         harness.snapshot(i, ++timestamp);
@@ -233,7 +236,8 @@ public void testOrderedEventsBetweenCheckpoints() throws Exception {
 
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -243,21 +247,21 @@ public void testOrderedEventsBetweenCheckpoints() throws Exception {
       RowData row1 = SimpleDataUtil.createRowData(1, "hello");
       DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
 
-      harness.processElement(of(dataFile1), ++timestamp);
+      long firstCheckpointId = 1;
+      harness.processElement(of(firstCheckpointId, dataFile1), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 1. snapshotState for checkpoint#1
-      long firstCheckpointId = 1;
       harness.snapshot(firstCheckpointId, ++timestamp);
       assertFlinkManifests(1);
 
       RowData row2 = SimpleDataUtil.createRowData(2, "world");
       DataFile dataFile2 = writeDataFile("data-2", ImmutableList.of(row2));
-      harness.processElement(of(dataFile2), ++timestamp);
+      long secondCheckpointId = 2;
+      harness.processElement(of(secondCheckpointId, dataFile2), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 2. snapshotState for checkpoint#2
-      long secondCheckpointId = 2;
       harness.snapshot(secondCheckpointId, ++timestamp);
       assertFlinkManifests(2);
 
@@ -286,7 +290,8 @@ public void testDisorderedEventsBetweenCheckpoints() throws Exception {
 
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -296,21 +301,21 @@ public void testDisorderedEventsBetweenCheckpoints() throws Exception {
       RowData row1 = SimpleDataUtil.createRowData(1, "hello");
       DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
 
-      harness.processElement(of(dataFile1), ++timestamp);
+      long firstCheckpointId = 1;
+      harness.processElement(of(firstCheckpointId, dataFile1), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 1. snapshotState for checkpoint#1
-      long firstCheckpointId = 1;
       harness.snapshot(firstCheckpointId, ++timestamp);
       assertFlinkManifests(1);
 
       RowData row2 = SimpleDataUtil.createRowData(2, "world");
       DataFile dataFile2 = writeDataFile("data-2", ImmutableList.of(row2));
-      harness.processElement(of(dataFile2), ++timestamp);
+      long secondCheckpointId = 2;
+      harness.processElement(of(secondCheckpointId, dataFile2), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 2. snapshotState for checkpoint#2
-      long secondCheckpointId = 2;
       harness.snapshot(secondCheckpointId, ++timestamp);
       assertFlinkManifests(2);
 
@@ -337,7 +342,8 @@ public void testRecoveryFromValidSnapshot() throws Exception {
 
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -349,8 +355,8 @@ public void testRecoveryFromValidSnapshot() throws Exception {
       expectedRows.add(row);
       DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row));
 
-      harness.processElement(of(dataFile1), ++timestamp);
-      snapshot = harness.snapshot(++checkpointId, ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile1), ++timestamp);
+      snapshot = harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
 
       harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -362,7 +368,8 @@ public void testRecoveryFromValidSnapshot() throws Exception {
     }
 
     // Restore from the given snapshot
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.getStreamConfig().setOperatorID(operatorId);
       harness.setup();
       harness.initializeState(snapshot);
@@ -375,9 +382,9 @@ public void testRecoveryFromValidSnapshot() throws Exception {
       RowData row = SimpleDataUtil.createRowData(2, "world");
       expectedRows.add(row);
       DataFile dataFile = writeDataFile("data-2", ImmutableList.of(row));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
 
-      harness.snapshot(++checkpointId, ++timestamp);
+      harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
 
       harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -400,7 +407,8 @@ public void testRecoveryFromSnapshotWithoutCompletedNotification() throws Except
     List<RowData> expectedRows = Lists.newArrayList();
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -411,15 +419,16 @@ public void testRecoveryFromSnapshotWithoutCompletedNotification() throws Except
       RowData row = SimpleDataUtil.createRowData(1, "hello");
       expectedRows.add(row);
       DataFile dataFile = writeDataFile("data-1", ImmutableList.of(row));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
 
-      snapshot = harness.snapshot(++checkpointId, ++timestamp);
+      snapshot = harness.snapshot(checkpointId, ++timestamp);
       SimpleDataUtil.assertTableRows(table, ImmutableList.of(), branch);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
       assertFlinkManifests(1);
     }
 
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.getStreamConfig().setOperatorID(operatorId);
       harness.setup();
       harness.initializeState(snapshot);
@@ -446,15 +455,15 @@ public void testRecoveryFromSnapshotWithoutCompletedNotification() throws Except
       RowData row = SimpleDataUtil.createRowData(2, "world");
       expectedRows.add(row);
       DataFile dataFile = writeDataFile("data-2", ImmutableList.of(row));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
 
-      snapshot = harness.snapshot(++checkpointId, ++timestamp);
+      snapshot = harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
     }
 
     // Redeploying flink job from external checkpoint.
     JobID newJobId = new JobID();
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
         createStreamSink(newJobId)) {
       harness.setup();
       harness.initializeState(snapshot);
@@ -473,9 +482,9 @@ public void testRecoveryFromSnapshotWithoutCompletedNotification() throws Except
       RowData row = SimpleDataUtil.createRowData(3, "foo");
       expectedRows.add(row);
       DataFile dataFile = writeDataFile("data-3", ImmutableList.of(row));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
 
-      harness.snapshot(++checkpointId, ++timestamp);
+      harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
 
       harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -496,7 +505,7 @@ public void testStartAnotherJobToWriteSameTable() throws Exception {
 
     JobID oldJobId = new JobID();
     OperatorID oldOperatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
         createStreamSink(oldJobId)) {
       harness.setup();
       harness.open();
@@ -510,8 +519,8 @@ public void testStartAnotherJobToWriteSameTable() throws Exception {
         tableRows.addAll(rows);
 
         DataFile dataFile = writeDataFile(String.format("data-%d", i), rows);
-        harness.processElement(of(dataFile), ++timestamp);
-        harness.snapshot(++checkpointId, ++timestamp);
+        harness.processElement(of(++checkpointId, dataFile), ++timestamp);
+        harness.snapshot(checkpointId, ++timestamp);
         assertFlinkManifests(1);
 
         harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -528,7 +537,7 @@ public void testStartAnotherJobToWriteSameTable() throws Exception {
     timestamp = 0;
     JobID newJobId = new JobID();
     OperatorID newOperatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
         createStreamSink(newJobId)) {
       harness.setup();
       harness.open();
@@ -542,8 +551,8 @@ public void testStartAnotherJobToWriteSameTable() throws Exception {
       tableRows.addAll(rows);
 
       DataFile dataFile = writeDataFile("data-new-1", rows);
-      harness.processElement(of(dataFile), ++timestamp);
-      harness.snapshot(++checkpointId, ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
+      harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
 
       harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -567,7 +576,8 @@ public void testMultipleJobsWriteSameTable() throws Exception {
       int checkpointId = i / 3;
       JobID jobId = jobs[jobIndex];
       OperatorID operatorId = operatorIds[jobIndex];
-      try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+      try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+          createStreamSink(jobId)) {
         harness.getStreamConfig().setOperatorID(operatorId);
         harness.setup();
         harness.open();
@@ -579,7 +589,7 @@ public void testMultipleJobsWriteSameTable() throws Exception {
         tableRows.addAll(rows);
 
         DataFile dataFile = writeDataFile(String.format("data-%d", i), rows);
-        harness.processElement(of(dataFile), ++timestamp);
+        harness.processElement(of(checkpointId + 1, dataFile), ++timestamp);
         harness.snapshot(checkpointId + 1, ++timestamp);
         assertFlinkManifests(1);
 
@@ -603,8 +613,10 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
     JobID jobId = new JobID();
     OperatorID operatorId1 = new OperatorID();
     OperatorID operatorId2 = new OperatorID();
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 = createStreamSink(jobId);
-        OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness1 =
+            createStreamSink(jobId);
+        OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness2 =
+            createStreamSink(jobId)) {
       harness1.getStreamConfig().setOperatorID(operatorId1);
       harness1.setup();
       harness1.open();
@@ -620,14 +632,14 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
       expectedRows.add(row1);
       DataFile dataFile1 = writeDataFile("data-1-1", ImmutableList.of(row1));
 
-      harness1.processElement(of(dataFile1), ++timestamp);
-      snapshot1 = harness1.snapshot(++checkpointId, ++timestamp);
+      harness1.processElement(of(++checkpointId, dataFile1), ++timestamp);
+      snapshot1 = harness1.snapshot(checkpointId, ++timestamp);
 
       RowData row2 = SimpleDataUtil.createRowData(1, "hello2");
       expectedRows.add(row2);
       DataFile dataFile2 = writeDataFile("data-1-2", ImmutableList.of(row2));
 
-      harness2.processElement(of(dataFile2), ++timestamp);
+      harness2.processElement(of(checkpointId, dataFile2), ++timestamp);
       snapshot2 = harness2.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(2);
 
@@ -643,8 +655,10 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
     }
 
     // Restore from the given snapshot
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 = createStreamSink(jobId);
-        OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness1 =
+            createStreamSink(jobId);
+        OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness2 =
+            createStreamSink(jobId)) {
       harness1.getStreamConfig().setOperatorID(operatorId1);
       harness1.setup();
       harness1.initializeState(snapshot1);
@@ -668,13 +682,13 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
       expectedRows.add(row1);
       DataFile dataFile1 = writeDataFile("data-2-1", ImmutableList.of(row1));
 
-      harness1.processElement(of(dataFile1), ++timestamp);
-      harness1.snapshot(++checkpointId, ++timestamp);
+      harness1.processElement(of(++checkpointId, dataFile1), ++timestamp);
+      harness1.snapshot(checkpointId, ++timestamp);
 
       RowData row2 = SimpleDataUtil.createRowData(2, "world2");
       expectedRows.add(row2);
       DataFile dataFile2 = writeDataFile("data-2-2", ImmutableList.of(row2));
-      harness2.processElement(of(dataFile2), ++timestamp);
+      harness2.processElement(of(checkpointId, dataFile2), ++timestamp);
       harness2.snapshot(checkpointId, ++timestamp);
 
       assertFlinkManifests(2);
@@ -694,7 +708,8 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
   public void testBoundedStream() throws Exception {
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -706,13 +721,14 @@ public void testBoundedStream() throws Exception {
       List<RowData> tableRows = Lists.newArrayList(SimpleDataUtil.createRowData(1, "word-1"));
 
       DataFile dataFile = writeDataFile("data-1", tableRows);
-      harness.processElement(of(dataFile), 1);
+      harness.processElement(of(IcebergStreamWriter.END_INPUT_CHECKPOINT_ID, dataFile), 1);
       ((BoundedOneInput) harness.getOneInputOperator()).endInput();
 
       assertFlinkManifests(0);
       SimpleDataUtil.assertTableRows(table, tableRows, branch);
       assertSnapshotSize(1);
-      assertMaxCommittedCheckpointId(jobId, operatorId, Long.MAX_VALUE);
+      assertMaxCommittedCheckpointId(
+          jobId, operatorId, IcebergStreamWriter.END_INPUT_CHECKPOINT_ID);
       assertThat(SimpleDataUtil.latestSnapshot(table, branch).summary())
           .containsEntry("flink.test", TestIcebergFilesCommitter.class.getName());
     }
@@ -725,7 +741,8 @@ public void testFlinkManifests() throws Exception {
 
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -735,7 +752,7 @@ public void testFlinkManifests() throws Exception {
       RowData row1 = SimpleDataUtil.createRowData(1, "hello");
       DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
 
-      harness.processElement(of(dataFile1), ++timestamp);
+      harness.processElement(of(checkpoint, dataFile1), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 1. snapshotState for checkpoint#1
@@ -775,7 +792,8 @@ public void testDeleteFiles() throws Exception {
     OperatorID operatorId;
     FileAppenderFactory<RowData> appenderFactory = createDeletableAppenderFactory();
 
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -784,7 +802,7 @@ public void testDeleteFiles() throws Exception {
 
       RowData row1 = SimpleDataUtil.createInsert(1, "aaa");
       DataFile dataFile1 = writeDataFile("data-file-1", ImmutableList.of(row1));
-      harness.processElement(of(dataFile1), ++timestamp);
+      harness.processElement(of(checkpoint, dataFile1), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 1. snapshotState for checkpoint#1
@@ -816,13 +834,15 @@ public void testDeleteFiles() throws Exception {
       RowData delete1 = SimpleDataUtil.createDelete(1, "aaa");
       DeleteFile deleteFile1 =
           writeEqDeleteFile(appenderFactory, "delete-file-1", ImmutableList.of(delete1));
+      assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
       harness.processElement(
-          WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile1).build(),
+          new FlinkWriteResult(
+              ++checkpoint,
+              WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile1).build()),
           ++timestamp);
-      assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
 
       // 5. snapshotState for checkpoint#2
-      harness.snapshot(++checkpoint, ++timestamp);
+      harness.snapshot(checkpoint, ++timestamp);
       assertFlinkManifests(2);
 
       // 6. notifyCheckpointComplete for checkpoint#2
@@ -846,7 +866,8 @@ public void testCommitTwoCheckpointsInSingleTxn() throws Exception {
     OperatorID operatorId;
     FileAppenderFactory<RowData> appenderFactory = createDeletableAppenderFactory();
 
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -860,7 +881,9 @@ public void testCommitTwoCheckpointsInSingleTxn() throws Exception {
       DeleteFile deleteFile1 =
           writeEqDeleteFile(appenderFactory, "delete-file-1", ImmutableList.of(delete3));
       harness.processElement(
-          WriteResult.builder().addDataFiles(dataFile1).addDeleteFiles(deleteFile1).build(),
+          new FlinkWriteResult(
+              checkpoint,
+              WriteResult.builder().addDataFiles(dataFile1).addDeleteFiles(deleteFile1).build()),
           ++timestamp);
 
       // The 1th snapshotState.
@@ -872,11 +895,13 @@ public void testCommitTwoCheckpointsInSingleTxn() throws Exception {
       DeleteFile deleteFile2 =
           writeEqDeleteFile(appenderFactory, "delete-file-2", ImmutableList.of(delete2));
       harness.processElement(
-          WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile2).build(),
+          new FlinkWriteResult(
+              ++checkpoint,
+              WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile2).build()),
           ++timestamp);
 
       // The 2nd snapshotState.
-      harness.snapshot(++checkpoint, ++timestamp);
+      harness.snapshot(checkpoint, ++timestamp);
 
       // Notify the 2nd snapshot to complete.
       harness.notifyOfCompletedCheckpoint(checkpoint);
@@ -887,6 +912,79 @@ public void testCommitTwoCheckpointsInSingleTxn() throws Exception {
     }
   }
 
+  /**
+   * The testcase is to simulate upserting to an Iceberg V2 table, and facing the following
+   * scenario:
+   *
+   * <ul>
+   *   <li>A specific row is updated
+   *   <li>The prepareSnapshotPreBarrier triggered
+   *   <li>Checkpoint failed for reasons outside of the Iceberg connector
+   *   <li>The specific row is updated again in the second checkpoint as well
+   *   <li>Second snapshot is triggered, and finished
+   * </ul>
+   *
+   * <p>Previously the files from the 2 snapshots were committed in a single Iceberg commit, as a
+   * results duplicate rows were created in the table.
+   *
+   * @throws Exception Exception
+   */
+  @TestTemplate
+  public void testCommitMultipleCheckpointsForV2Table() throws Exception {
+    assumeThat(formatVersion)
+        .as("Only support equality-delete in format v2 or later.")
+        .isGreaterThan(1);
+
+    long timestamp = 0;
+    long checkpoint = 10;
+
+    JobID jobId = new JobID();
+    OperatorID operatorId;
+
+    FileAppenderFactory<RowData> appenderFactory =
+        new FlinkAppenderFactory(
+            table,
+            table.schema(),
+            FlinkSchemaUtil.convert(table.schema()),
+            table.properties(),
+            table.spec(),
+            new int[] {table.schema().findField("id").fieldId()},
+            table.schema(),
+            null);
+
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
+      harness.setup();
+      harness.open();
+      operatorId = harness.getOperator().getOperatorID();
+
+      assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
+
+      RowData insert1 = null;
+      RowData insert2 = null;
+      for (int i = 1; i <= 3; i++) {
+        insert1 = SimpleDataUtil.createInsert(1, "aaa" + i);
+        insert2 = SimpleDataUtil.createInsert(2, "bbb" + i);
+        DataFile dataFile = writeDataFile("data-file-" + i, ImmutableList.of(insert1, insert2));
+        DeleteFile deleteFile =
+            writeEqDeleteFile(
+                appenderFactory, "delete-file-" + i, ImmutableList.of(insert1, insert2));
+        harness.processElement(
+            new FlinkWriteResult(
+                ++checkpoint,
+                WriteResult.builder().addDataFiles(dataFile).addDeleteFiles(deleteFile).build()),
+            ++timestamp);
+      }
+
+      harness.snapshot(checkpoint, ++timestamp);
+      harness.notifyOfCompletedCheckpoint(checkpoint);
+      SimpleDataUtil.assertTableRows(table, ImmutableList.of(insert1, insert2), branch);
+      assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
+      assertFlinkManifests(0);
+      assertThat(table.snapshots()).hasSize(3);
+    }
+  }
+
   @TestTemplate
   public void testSpecEvolution() throws Exception {
     long timestamp = 0;
@@ -899,7 +997,8 @@ public void testSpecEvolution() throws Exception {
     DataFile dataFile;
     int specId;
 
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -910,7 +1009,7 @@ public void testSpecEvolution() throws Exception {
       RowData rowData = SimpleDataUtil.createRowData(checkpointId, "hello" + checkpointId);
       // table unpartitioned
       dataFile = writeDataFile("data-" + checkpointId, ImmutableList.of(rowData));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(checkpointId, dataFile), ++timestamp);
       rows.add(rowData);
       harness.snapshot(checkpointId, ++timestamp);
 
@@ -929,7 +1028,7 @@ public void testSpecEvolution() throws Exception {
       rowData = SimpleDataUtil.createRowData(checkpointId, "hello" + checkpointId);
       // write data with old partition spec
       dataFile = writeDataFile("data-" + checkpointId, ImmutableList.of(rowData), oldSpec, null);
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(checkpointId, dataFile), ++timestamp);
       rows.add(rowData);
       snapshot = harness.snapshot(checkpointId, ++timestamp);
 
@@ -947,7 +1046,8 @@ public void testSpecEvolution() throws Exception {
     }
 
     // Restore from the given snapshot
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.getStreamConfig().setOperatorID(operatorId);
       harness.setup();
       harness.initializeState(snapshot);
@@ -963,7 +1063,7 @@ public void testSpecEvolution() throws Exception {
       partition.set(0, checkpointId);
       dataFile =
           writeDataFile("data-" + checkpointId, ImmutableList.of(row), table.spec(), partition);
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(checkpointId, dataFile), ++timestamp);
       rows.add(row);
       harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
@@ -1089,7 +1189,7 @@ private void assertSnapshotSize(int expectedSnapshotSize) {
     assertThat(table.snapshots()).hasSize(expectedSnapshotSize);
   }
 
-  private OneInputStreamOperatorTestHarness<WriteResult, Void> createStreamSink(JobID jobID)
+  private OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> createStreamSink(JobID jobID)
       throws Exception {
     TestOperatorFactory factory = TestOperatorFactory.of(table.location(), branch, table.spec());
     return new OneInputStreamOperatorTestHarness<>(factory, createEnvironment(jobID));
@@ -1109,7 +1209,7 @@ private static MockEnvironment createEnvironment(JobID jobID) {
   }
 
   private static class TestOperatorFactory extends AbstractStreamOperatorFactory<Void>
-      implements OneInputStreamOperatorFactory<WriteResult, Void> {
+      implements OneInputStreamOperatorFactory<FlinkWriteResult, Void> {
     private final String tablePath;
     private final String branch;
     private final PartitionSpec spec;
diff --git a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
index 50283f7ad215..e13721a9f170 100644
--- a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
+++ b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
@@ -28,6 +28,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 import org.apache.flink.streaming.api.operators.BoundedOneInput;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.table.api.DataTypes;
@@ -102,7 +103,7 @@ public void before() throws IOException {
   @TestTemplate
   public void testWritingTable() throws Exception {
     long checkpointId = 1L;
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       // The first checkpoint
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
@@ -111,7 +112,8 @@ public void testWritingTable() throws Exception {
 
       testHarness.prepareSnapshotPreBarrier(checkpointId);
       int expectedDataFiles = partitioned ? 2 : 1;
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
@@ -123,7 +125,8 @@ public void testWritingTable() throws Exception {
 
       testHarness.prepareSnapshotPreBarrier(checkpointId);
       expectedDataFiles = partitioned ? 4 : 2;
-      result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
@@ -148,14 +151,15 @@ public void testWritingTable() throws Exception {
   public void testSnapshotTwice() throws Exception {
     long checkpointId = 1;
     long timestamp = 1;
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), timestamp++);
       testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), timestamp);
 
       testHarness.prepareSnapshotPreBarrier(checkpointId++);
       int expectedDataFiles = partitioned ? 2 : 1;
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
@@ -163,7 +167,10 @@ public void testSnapshotTwice() throws Exception {
       for (int i = 0; i < 5; i++) {
         testHarness.prepareSnapshotPreBarrier(checkpointId++);
 
-        result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+        result =
+            WriteResult.builder()
+                .addAll(getWriteResults(testHarness.extractOutputValues()))
+                .build();
         assertThat(result.deleteFiles()).isEmpty();
         assertThat(result.dataFiles()).hasSize(expectedDataFiles);
       }
@@ -172,14 +179,14 @@ public void testSnapshotTwice() throws Exception {
 
   @TestTemplate
   public void testTableWithoutSnapshot() throws Exception {
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       assertThat(testHarness.extractOutputValues()).isEmpty();
     }
     // Even if we closed the iceberg stream writer, there's no orphan data file.
     assertThat(scanDataFiles()).isEmpty();
 
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
       // Still not emit the data file yet, because there is no checkpoint.
@@ -212,7 +219,7 @@ private Set<String> scanDataFiles() throws IOException {
 
   @TestTemplate
   public void testBoundedStreamCloseWithEmittingDataFiles() throws Exception {
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
       testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), 2);
@@ -221,13 +228,15 @@ public void testBoundedStreamCloseWithEmittingDataFiles() throws Exception {
       ((BoundedOneInput) testHarness.getOneInputOperator()).endInput();
 
       int expectedDataFiles = partitioned ? 2 : 1;
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
       ((BoundedOneInput) testHarness.getOneInputOperator()).endInput();
 
-      result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       // Datafiles should not be sent again
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -236,7 +245,7 @@ public void testBoundedStreamCloseWithEmittingDataFiles() throws Exception {
 
   @TestTemplate
   public void testBoundedStreamTriggeredEndInputBeforeTriggeringCheckpoint() throws Exception {
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
       testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), 2);
@@ -244,13 +253,15 @@ public void testBoundedStreamTriggeredEndInputBeforeTriggeringCheckpoint() throw
       testHarness.endInput();
 
       int expectedDataFiles = partitioned ? 2 : 1;
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
       testHarness.prepareSnapshotPreBarrier(1L);
 
-      result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       // It should be ensured that after endInput is triggered, when prepareSnapshotPreBarrier
       // is triggered, write should only send WriteResult once
@@ -275,7 +286,7 @@ public void testTableWithTargetFileSize() throws Exception {
       }
     }
 
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       for (RowData row : rows) {
         testHarness.processElement(row, 1);
@@ -283,7 +294,8 @@ public void testTableWithTargetFileSize() throws Exception {
 
       // snapshot the operator.
       testHarness.prepareSnapshotPreBarrier(1);
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(8);
 
@@ -346,13 +358,14 @@ public void testPromotedFlinkDataType() throws Exception {
             record.copy(ImmutableMap.of("tinyint", 2, "smallint", 0, "int", 102)),
             record.copy(ImmutableMap.of("tinyint", 3, "smallint", 32767, "int", 103)));
 
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter(icebergTable, flinkSchema)) {
       for (RowData row : rows) {
         testHarness.processElement(row, 1);
       }
       testHarness.prepareSnapshotPreBarrier(1);
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(partitioned ? 3 : 1);
 
@@ -365,12 +378,18 @@ public void testPromotedFlinkDataType() throws Exception {
     SimpleDataUtil.assertTableRecords(location, expected);
   }
 
-  private OneInputStreamOperatorTestHarness<RowData, WriteResult> createIcebergStreamWriter()
+  private static List<WriteResult> getWriteResults(List<FlinkWriteResult> flinkWriteResults) {
+    return flinkWriteResults.stream()
+        .map(FlinkWriteResult::writeResult)
+        .collect(Collectors.toList());
+  }
+
+  private OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> createIcebergStreamWriter()
       throws Exception {
     return createIcebergStreamWriter(table, SimpleDataUtil.FLINK_SCHEMA);
   }
 
-  private OneInputStreamOperatorTestHarness<RowData, WriteResult> createIcebergStreamWriter(
+  private OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> createIcebergStreamWriter(
       Table icebergTable, TableSchema flinkSchema) throws Exception {
     RowType flinkRowType = FlinkSink.toFlinkRowType(icebergTable.schema(), flinkSchema);
     FlinkWriteConf flinkWriteConfig =
@@ -379,7 +398,7 @@ private OneInputStreamOperatorTestHarness<RowData, WriteResult> createIcebergStr
 
     IcebergStreamWriter<RowData> streamWriter =
         FlinkSink.createStreamWriter(() -> icebergTable, flinkWriteConfig, flinkRowType, null);
-    OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+    OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
         new OneInputStreamOperatorTestHarness<>(streamWriter, 1, 1, 0);
 
     harness.setup();
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
index 8faae1b05a4e..3299e7a97776 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
@@ -40,7 +40,6 @@
 import org.apache.iceberg.flink.SimpleDataUtil;
 import org.apache.iceberg.io.BaseTaskWriter;
 import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestTemplate;
@@ -210,8 +209,10 @@ public void testCompressionOrc() throws Exception {
         .containsEntry(TableProperties.ORC_COMPRESSION_STRATEGY, "speed");
   }
 
-  private static OneInputStreamOperatorTestHarness<RowData, WriteResult> createIcebergStreamWriter(
-      Table icebergTable, TableSchema flinkSchema, Map<String, String> override) throws Exception {
+  private static OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
+      createIcebergStreamWriter(
+          Table icebergTable, TableSchema flinkSchema, Map<String, String> override)
+          throws Exception {
     RowType flinkRowType = FlinkSink.toFlinkRowType(icebergTable.schema(), flinkSchema);
     FlinkWriteConf flinkWriteConfig =
         new FlinkWriteConf(
@@ -219,7 +220,7 @@ private static OneInputStreamOperatorTestHarness<RowData, WriteResult> createIce
 
     IcebergStreamWriter<RowData> streamWriter =
         FlinkSink.createStreamWriter(() -> icebergTable, flinkWriteConfig, flinkRowType, null);
-    OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+    OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
         new OneInputStreamOperatorTestHarness<>(streamWriter, 1, 1, 0);
 
     harness.setup();
@@ -230,7 +231,7 @@ private static OneInputStreamOperatorTestHarness<RowData, WriteResult> createIce
 
   private static Map<String, String> appenderProperties(
       Table table, TableSchema schema, Map<String, String> override) throws Exception {
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter(table, schema, override)) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
 
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
index 67ca9d08b206..7808771d9887 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
@@ -129,7 +129,8 @@ public void testCommitTxnWithoutDataFiles() throws Exception {
     long timestamp = 0;
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -160,7 +161,8 @@ public void testMaxContinuousEmptyCommits() throws Exception {
     JobID jobId = new JobID();
     long checkpointId = 0;
     long timestamp = 0;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
 
@@ -175,8 +177,8 @@ public void testMaxContinuousEmptyCommits() throws Exception {
     }
   }
 
-  private WriteResult of(DataFile dataFile) {
-    return WriteResult.builder().addDataFiles(dataFile).build();
+  private FlinkWriteResult of(long checkpointId, DataFile dataFile) {
+    return new FlinkWriteResult(checkpointId, WriteResult.builder().addDataFiles(dataFile).build());
   }
 
   @TestTemplate
@@ -192,7 +194,8 @@ public void testCommitTxn() throws Exception {
 
     JobID jobID = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobID)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobID)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -203,7 +206,7 @@ public void testCommitTxn() throws Exception {
       for (int i = 1; i <= 3; i++) {
         RowData rowData = SimpleDataUtil.createRowData(i, "hello" + i);
         DataFile dataFile = writeDataFile("data-" + i, ImmutableList.of(rowData));
-        harness.processElement(of(dataFile), ++timestamp);
+        harness.processElement(of(i, dataFile), ++timestamp);
         rows.add(rowData);
 
         harness.snapshot(i, ++timestamp);
@@ -232,7 +235,8 @@ public void testOrderedEventsBetweenCheckpoints() throws Exception {
 
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -242,21 +246,21 @@ public void testOrderedEventsBetweenCheckpoints() throws Exception {
       RowData row1 = SimpleDataUtil.createRowData(1, "hello");
       DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
 
-      harness.processElement(of(dataFile1), ++timestamp);
+      long firstCheckpointId = 1;
+      harness.processElement(of(firstCheckpointId, dataFile1), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 1. snapshotState for checkpoint#1
-      long firstCheckpointId = 1;
       harness.snapshot(firstCheckpointId, ++timestamp);
       assertFlinkManifests(1);
 
       RowData row2 = SimpleDataUtil.createRowData(2, "world");
       DataFile dataFile2 = writeDataFile("data-2", ImmutableList.of(row2));
-      harness.processElement(of(dataFile2), ++timestamp);
+      long secondCheckpointId = 2;
+      harness.processElement(of(secondCheckpointId, dataFile2), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 2. snapshotState for checkpoint#2
-      long secondCheckpointId = 2;
       harness.snapshot(secondCheckpointId, ++timestamp);
       assertFlinkManifests(2);
 
@@ -285,7 +289,8 @@ public void testDisorderedEventsBetweenCheckpoints() throws Exception {
 
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -295,21 +300,21 @@ public void testDisorderedEventsBetweenCheckpoints() throws Exception {
       RowData row1 = SimpleDataUtil.createRowData(1, "hello");
       DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
 
-      harness.processElement(of(dataFile1), ++timestamp);
+      long firstCheckpointId = 1;
+      harness.processElement(of(firstCheckpointId, dataFile1), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 1. snapshotState for checkpoint#1
-      long firstCheckpointId = 1;
       harness.snapshot(firstCheckpointId, ++timestamp);
       assertFlinkManifests(1);
 
       RowData row2 = SimpleDataUtil.createRowData(2, "world");
       DataFile dataFile2 = writeDataFile("data-2", ImmutableList.of(row2));
-      harness.processElement(of(dataFile2), ++timestamp);
+      long secondCheckpointId = 2;
+      harness.processElement(of(secondCheckpointId, dataFile2), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 2. snapshotState for checkpoint#2
-      long secondCheckpointId = 2;
       harness.snapshot(secondCheckpointId, ++timestamp);
       assertFlinkManifests(2);
 
@@ -336,7 +341,8 @@ public void testRecoveryFromValidSnapshot() throws Exception {
 
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -348,8 +354,8 @@ public void testRecoveryFromValidSnapshot() throws Exception {
       expectedRows.add(row);
       DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row));
 
-      harness.processElement(of(dataFile1), ++timestamp);
-      snapshot = harness.snapshot(++checkpointId, ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile1), ++timestamp);
+      snapshot = harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
 
       harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -361,7 +367,8 @@ public void testRecoveryFromValidSnapshot() throws Exception {
     }
 
     // Restore from the given snapshot
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.getStreamConfig().setOperatorID(operatorId);
       harness.setup();
       harness.initializeState(snapshot);
@@ -374,9 +381,9 @@ public void testRecoveryFromValidSnapshot() throws Exception {
       RowData row = SimpleDataUtil.createRowData(2, "world");
       expectedRows.add(row);
       DataFile dataFile = writeDataFile("data-2", ImmutableList.of(row));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
 
-      harness.snapshot(++checkpointId, ++timestamp);
+      harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
 
       harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -399,7 +406,8 @@ public void testRecoveryFromSnapshotWithoutCompletedNotification() throws Except
     List<RowData> expectedRows = Lists.newArrayList();
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -410,15 +418,16 @@ public void testRecoveryFromSnapshotWithoutCompletedNotification() throws Except
       RowData row = SimpleDataUtil.createRowData(1, "hello");
       expectedRows.add(row);
       DataFile dataFile = writeDataFile("data-1", ImmutableList.of(row));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
 
-      snapshot = harness.snapshot(++checkpointId, ++timestamp);
+      snapshot = harness.snapshot(checkpointId, ++timestamp);
       SimpleDataUtil.assertTableRows(table, ImmutableList.of(), branch);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
       assertFlinkManifests(1);
     }
 
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.getStreamConfig().setOperatorID(operatorId);
       harness.setup();
       harness.initializeState(snapshot);
@@ -445,15 +454,15 @@ public void testRecoveryFromSnapshotWithoutCompletedNotification() throws Except
       RowData row = SimpleDataUtil.createRowData(2, "world");
       expectedRows.add(row);
       DataFile dataFile = writeDataFile("data-2", ImmutableList.of(row));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
 
-      snapshot = harness.snapshot(++checkpointId, ++timestamp);
+      snapshot = harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
     }
 
     // Redeploying flink job from external checkpoint.
     JobID newJobId = new JobID();
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
         createStreamSink(newJobId)) {
       harness.setup();
       harness.initializeState(snapshot);
@@ -472,9 +481,9 @@ public void testRecoveryFromSnapshotWithoutCompletedNotification() throws Except
       RowData row = SimpleDataUtil.createRowData(3, "foo");
       expectedRows.add(row);
       DataFile dataFile = writeDataFile("data-3", ImmutableList.of(row));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
 
-      harness.snapshot(++checkpointId, ++timestamp);
+      harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
 
       harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -495,7 +504,7 @@ public void testStartAnotherJobToWriteSameTable() throws Exception {
 
     JobID oldJobId = new JobID();
     OperatorID oldOperatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
         createStreamSink(oldJobId)) {
       harness.setup();
       harness.open();
@@ -509,8 +518,8 @@ public void testStartAnotherJobToWriteSameTable() throws Exception {
         tableRows.addAll(rows);
 
         DataFile dataFile = writeDataFile(String.format("data-%d", i), rows);
-        harness.processElement(of(dataFile), ++timestamp);
-        harness.snapshot(++checkpointId, ++timestamp);
+        harness.processElement(of(++checkpointId, dataFile), ++timestamp);
+        harness.snapshot(checkpointId, ++timestamp);
         assertFlinkManifests(1);
 
         harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -527,7 +536,7 @@ public void testStartAnotherJobToWriteSameTable() throws Exception {
     timestamp = 0;
     JobID newJobId = new JobID();
     OperatorID newOperatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
         createStreamSink(newJobId)) {
       harness.setup();
       harness.open();
@@ -541,8 +550,8 @@ public void testStartAnotherJobToWriteSameTable() throws Exception {
       tableRows.addAll(rows);
 
       DataFile dataFile = writeDataFile("data-new-1", rows);
-      harness.processElement(of(dataFile), ++timestamp);
-      harness.snapshot(++checkpointId, ++timestamp);
+      harness.processElement(of(++checkpointId, dataFile), ++timestamp);
+      harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
 
       harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -566,7 +575,8 @@ public void testMultipleJobsWriteSameTable() throws Exception {
       int checkpointId = i / 3;
       JobID jobId = jobs[jobIndex];
       OperatorID operatorId = operatorIds[jobIndex];
-      try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+      try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+          createStreamSink(jobId)) {
         harness.getStreamConfig().setOperatorID(operatorId);
         harness.setup();
         harness.open();
@@ -578,7 +588,7 @@ public void testMultipleJobsWriteSameTable() throws Exception {
         tableRows.addAll(rows);
 
         DataFile dataFile = writeDataFile(String.format("data-%d", i), rows);
-        harness.processElement(of(dataFile), ++timestamp);
+        harness.processElement(of(checkpointId + 1, dataFile), ++timestamp);
         harness.snapshot(checkpointId + 1, ++timestamp);
         assertFlinkManifests(1);
 
@@ -602,8 +612,10 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
     JobID jobId = new JobID();
     OperatorID operatorId1 = new OperatorID();
     OperatorID operatorId2 = new OperatorID();
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 = createStreamSink(jobId);
-        OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness1 =
+            createStreamSink(jobId);
+        OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness2 =
+            createStreamSink(jobId)) {
       harness1.getStreamConfig().setOperatorID(operatorId1);
       harness1.setup();
       harness1.open();
@@ -619,14 +631,14 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
       expectedRows.add(row1);
       DataFile dataFile1 = writeDataFile("data-1-1", ImmutableList.of(row1));
 
-      harness1.processElement(of(dataFile1), ++timestamp);
-      snapshot1 = harness1.snapshot(++checkpointId, ++timestamp);
+      harness1.processElement(of(++checkpointId, dataFile1), ++timestamp);
+      snapshot1 = harness1.snapshot(checkpointId, ++timestamp);
 
       RowData row2 = SimpleDataUtil.createRowData(1, "hello2");
       expectedRows.add(row2);
       DataFile dataFile2 = writeDataFile("data-1-2", ImmutableList.of(row2));
 
-      harness2.processElement(of(dataFile2), ++timestamp);
+      harness2.processElement(of(checkpointId, dataFile2), ++timestamp);
       snapshot2 = harness2.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(2);
 
@@ -642,8 +654,10 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
     }
 
     // Restore from the given snapshot
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 = createStreamSink(jobId);
-        OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness1 =
+            createStreamSink(jobId);
+        OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness2 =
+            createStreamSink(jobId)) {
       harness1.getStreamConfig().setOperatorID(operatorId1);
       harness1.setup();
       harness1.initializeState(snapshot1);
@@ -667,13 +681,13 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
       expectedRows.add(row1);
       DataFile dataFile1 = writeDataFile("data-2-1", ImmutableList.of(row1));
 
-      harness1.processElement(of(dataFile1), ++timestamp);
-      harness1.snapshot(++checkpointId, ++timestamp);
+      harness1.processElement(of(++checkpointId, dataFile1), ++timestamp);
+      harness1.snapshot(checkpointId, ++timestamp);
 
       RowData row2 = SimpleDataUtil.createRowData(2, "world2");
       expectedRows.add(row2);
       DataFile dataFile2 = writeDataFile("data-2-2", ImmutableList.of(row2));
-      harness2.processElement(of(dataFile2), ++timestamp);
+      harness2.processElement(of(checkpointId, dataFile2), ++timestamp);
       harness2.snapshot(checkpointId, ++timestamp);
 
       assertFlinkManifests(2);
@@ -693,7 +707,8 @@ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
   public void testBoundedStream() throws Exception {
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -705,13 +720,14 @@ public void testBoundedStream() throws Exception {
       List<RowData> tableRows = Lists.newArrayList(SimpleDataUtil.createRowData(1, "word-1"));
 
       DataFile dataFile = writeDataFile("data-1", tableRows);
-      harness.processElement(of(dataFile), 1);
+      harness.processElement(of(IcebergStreamWriter.END_INPUT_CHECKPOINT_ID, dataFile), 1);
       ((BoundedOneInput) harness.getOneInputOperator()).endInput();
 
       assertFlinkManifests(0);
       SimpleDataUtil.assertTableRows(table, tableRows, branch);
       assertSnapshotSize(1);
-      assertMaxCommittedCheckpointId(jobId, operatorId, Long.MAX_VALUE);
+      assertMaxCommittedCheckpointId(
+          jobId, operatorId, IcebergStreamWriter.END_INPUT_CHECKPOINT_ID);
       assertThat(SimpleDataUtil.latestSnapshot(table, branch).summary())
           .containsEntry("flink.test", TestIcebergFilesCommitter.class.getName());
     }
@@ -724,7 +740,8 @@ public void testFlinkManifests() throws Exception {
 
     JobID jobId = new JobID();
     OperatorID operatorId;
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -734,7 +751,7 @@ public void testFlinkManifests() throws Exception {
       RowData row1 = SimpleDataUtil.createRowData(1, "hello");
       DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
 
-      harness.processElement(of(dataFile1), ++timestamp);
+      harness.processElement(of(checkpoint, dataFile1), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 1. snapshotState for checkpoint#1
@@ -774,7 +791,8 @@ public void testDeleteFiles() throws Exception {
     OperatorID operatorId;
     FileAppenderFactory<RowData> appenderFactory = createDeletableAppenderFactory();
 
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -783,7 +801,7 @@ public void testDeleteFiles() throws Exception {
 
       RowData row1 = SimpleDataUtil.createInsert(1, "aaa");
       DataFile dataFile1 = writeDataFile("data-file-1", ImmutableList.of(row1));
-      harness.processElement(of(dataFile1), ++timestamp);
+      harness.processElement(of(checkpoint, dataFile1), ++timestamp);
       assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
 
       // 1. snapshotState for checkpoint#1
@@ -815,13 +833,15 @@ public void testDeleteFiles() throws Exception {
       RowData delete1 = SimpleDataUtil.createDelete(1, "aaa");
       DeleteFile deleteFile1 =
           writeEqDeleteFile(appenderFactory, "delete-file-1", ImmutableList.of(delete1));
+      assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
       harness.processElement(
-          WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile1).build(),
+          new FlinkWriteResult(
+              ++checkpoint,
+              WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile1).build()),
           ++timestamp);
-      assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
 
       // 5. snapshotState for checkpoint#2
-      harness.snapshot(++checkpoint, ++timestamp);
+      harness.snapshot(checkpoint, ++timestamp);
       assertFlinkManifests(2);
 
       // 6. notifyCheckpointComplete for checkpoint#2
@@ -845,7 +865,8 @@ public void testCommitTwoCheckpointsInSingleTxn() throws Exception {
     OperatorID operatorId;
     FileAppenderFactory<RowData> appenderFactory = createDeletableAppenderFactory();
 
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -859,7 +880,9 @@ public void testCommitTwoCheckpointsInSingleTxn() throws Exception {
       DeleteFile deleteFile1 =
           writeEqDeleteFile(appenderFactory, "delete-file-1", ImmutableList.of(delete3));
       harness.processElement(
-          WriteResult.builder().addDataFiles(dataFile1).addDeleteFiles(deleteFile1).build(),
+          new FlinkWriteResult(
+              checkpoint,
+              WriteResult.builder().addDataFiles(dataFile1).addDeleteFiles(deleteFile1).build()),
           ++timestamp);
 
       // The 1th snapshotState.
@@ -871,11 +894,13 @@ public void testCommitTwoCheckpointsInSingleTxn() throws Exception {
       DeleteFile deleteFile2 =
           writeEqDeleteFile(appenderFactory, "delete-file-2", ImmutableList.of(delete2));
       harness.processElement(
-          WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile2).build(),
+          new FlinkWriteResult(
+              ++checkpoint,
+              WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile2).build()),
           ++timestamp);
 
       // The 2nd snapshotState.
-      harness.snapshot(++checkpoint, ++timestamp);
+      harness.snapshot(checkpoint, ++timestamp);
 
       // Notify the 2nd snapshot to complete.
       harness.notifyOfCompletedCheckpoint(checkpoint);
@@ -886,6 +911,79 @@ public void testCommitTwoCheckpointsInSingleTxn() throws Exception {
     }
   }
 
+  /**
+   * The testcase is to simulate upserting to an Iceberg V2 table, and facing the following
+   * scenario:
+   *
+   * <ul>
+   *   <li>A specific row is updated
+   *   <li>The prepareSnapshotPreBarrier triggered
+   *   <li>Checkpoint failed for reasons outside of the Iceberg connector
+   *   <li>The specific row is updated again in the second checkpoint as well
+   *   <li>Second snapshot is triggered, and finished
+   * </ul>
+   *
+   * <p>Previously the files from the 2 snapshots were committed in a single Iceberg commit, as a
+   * results duplicate rows were created in the table.
+   *
+   * @throws Exception Exception
+   */
+  @TestTemplate
+  public void testCommitMultipleCheckpointsForV2Table() throws Exception {
+    assumeThat(formatVersion)
+        .as("Only support equality-delete in format v2 or later.")
+        .isGreaterThan(1);
+
+    long timestamp = 0;
+    long checkpoint = 10;
+
+    JobID jobId = new JobID();
+    OperatorID operatorId;
+
+    FileAppenderFactory<RowData> appenderFactory =
+        new FlinkAppenderFactory(
+            table,
+            table.schema(),
+            FlinkSchemaUtil.convert(table.schema()),
+            table.properties(),
+            table.spec(),
+            new int[] {table.schema().findField("id").fieldId()},
+            table.schema(),
+            null);
+
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
+      harness.setup();
+      harness.open();
+      operatorId = harness.getOperator().getOperatorID();
+
+      assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
+
+      RowData insert1 = null;
+      RowData insert2 = null;
+      for (int i = 1; i <= 3; i++) {
+        insert1 = SimpleDataUtil.createInsert(1, "aaa" + i);
+        insert2 = SimpleDataUtil.createInsert(2, "bbb" + i);
+        DataFile dataFile = writeDataFile("data-file-" + i, ImmutableList.of(insert1, insert2));
+        DeleteFile deleteFile =
+            writeEqDeleteFile(
+                appenderFactory, "delete-file-" + i, ImmutableList.of(insert1, insert2));
+        harness.processElement(
+            new FlinkWriteResult(
+                ++checkpoint,
+                WriteResult.builder().addDataFiles(dataFile).addDeleteFiles(deleteFile).build()),
+            ++timestamp);
+      }
+
+      harness.snapshot(checkpoint, ++timestamp);
+      harness.notifyOfCompletedCheckpoint(checkpoint);
+      SimpleDataUtil.assertTableRows(table, ImmutableList.of(insert1, insert2), branch);
+      assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
+      assertFlinkManifests(0);
+      assertThat(table.snapshots()).hasSize(3);
+    }
+  }
+
   @TestTemplate
   public void testSpecEvolution() throws Exception {
     long timestamp = 0;
@@ -898,7 +996,8 @@ public void testSpecEvolution() throws Exception {
     DataFile dataFile;
     int specId;
 
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.setup();
       harness.open();
       operatorId = harness.getOperator().getOperatorID();
@@ -909,7 +1008,7 @@ public void testSpecEvolution() throws Exception {
       RowData rowData = SimpleDataUtil.createRowData(checkpointId, "hello" + checkpointId);
       // table unpartitioned
       dataFile = writeDataFile("data-" + checkpointId, ImmutableList.of(rowData));
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(checkpointId, dataFile), ++timestamp);
       rows.add(rowData);
       harness.snapshot(checkpointId, ++timestamp);
 
@@ -928,7 +1027,7 @@ public void testSpecEvolution() throws Exception {
       rowData = SimpleDataUtil.createRowData(checkpointId, "hello" + checkpointId);
       // write data with old partition spec
       dataFile = writeDataFile("data-" + checkpointId, ImmutableList.of(rowData), oldSpec, null);
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(checkpointId, dataFile), ++timestamp);
       rows.add(rowData);
       snapshot = harness.snapshot(checkpointId, ++timestamp);
 
@@ -946,7 +1045,8 @@ public void testSpecEvolution() throws Exception {
     }
 
     // Restore from the given snapshot
-    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
+    try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+        createStreamSink(jobId)) {
       harness.getStreamConfig().setOperatorID(operatorId);
       harness.setup();
       harness.initializeState(snapshot);
@@ -962,7 +1062,7 @@ public void testSpecEvolution() throws Exception {
       partition.set(0, checkpointId);
       dataFile =
           writeDataFile("data-" + checkpointId, ImmutableList.of(row), table.spec(), partition);
-      harness.processElement(of(dataFile), ++timestamp);
+      harness.processElement(of(checkpointId, dataFile), ++timestamp);
       rows.add(row);
       harness.snapshot(checkpointId, ++timestamp);
       assertFlinkManifests(1);
@@ -1088,7 +1188,7 @@ private void assertSnapshotSize(int expectedSnapshotSize) {
     assertThat(table.snapshots()).hasSize(expectedSnapshotSize);
   }
 
-  private OneInputStreamOperatorTestHarness<WriteResult, Void> createStreamSink(JobID jobID)
+  private OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> createStreamSink(JobID jobID)
       throws Exception {
     TestOperatorFactory factory = TestOperatorFactory.of(table.location(), branch, table.spec());
     return new OneInputStreamOperatorTestHarness<>(factory, createEnvironment(jobID));
@@ -1108,7 +1208,7 @@ private static MockEnvironment createEnvironment(JobID jobID) {
   }
 
   private static class TestOperatorFactory extends AbstractStreamOperatorFactory<Void>
-      implements OneInputStreamOperatorFactory<WriteResult, Void> {
+      implements OneInputStreamOperatorFactory<FlinkWriteResult, Void> {
     private final String tablePath;
     private final String branch;
     private final PartitionSpec spec;
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
index 50283f7ad215..e13721a9f170 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
@@ -28,6 +28,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 import org.apache.flink.streaming.api.operators.BoundedOneInput;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.table.api.DataTypes;
@@ -102,7 +103,7 @@ public void before() throws IOException {
   @TestTemplate
   public void testWritingTable() throws Exception {
     long checkpointId = 1L;
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       // The first checkpoint
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
@@ -111,7 +112,8 @@ public void testWritingTable() throws Exception {
 
       testHarness.prepareSnapshotPreBarrier(checkpointId);
       int expectedDataFiles = partitioned ? 2 : 1;
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
@@ -123,7 +125,8 @@ public void testWritingTable() throws Exception {
 
       testHarness.prepareSnapshotPreBarrier(checkpointId);
       expectedDataFiles = partitioned ? 4 : 2;
-      result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
@@ -148,14 +151,15 @@ public void testWritingTable() throws Exception {
   public void testSnapshotTwice() throws Exception {
     long checkpointId = 1;
     long timestamp = 1;
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), timestamp++);
       testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), timestamp);
 
       testHarness.prepareSnapshotPreBarrier(checkpointId++);
       int expectedDataFiles = partitioned ? 2 : 1;
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
@@ -163,7 +167,10 @@ public void testSnapshotTwice() throws Exception {
       for (int i = 0; i < 5; i++) {
         testHarness.prepareSnapshotPreBarrier(checkpointId++);
 
-        result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+        result =
+            WriteResult.builder()
+                .addAll(getWriteResults(testHarness.extractOutputValues()))
+                .build();
         assertThat(result.deleteFiles()).isEmpty();
         assertThat(result.dataFiles()).hasSize(expectedDataFiles);
       }
@@ -172,14 +179,14 @@ public void testSnapshotTwice() throws Exception {
 
   @TestTemplate
   public void testTableWithoutSnapshot() throws Exception {
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       assertThat(testHarness.extractOutputValues()).isEmpty();
     }
     // Even if we closed the iceberg stream writer, there's no orphan data file.
     assertThat(scanDataFiles()).isEmpty();
 
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
       // Still not emit the data file yet, because there is no checkpoint.
@@ -212,7 +219,7 @@ private Set<String> scanDataFiles() throws IOException {
 
   @TestTemplate
   public void testBoundedStreamCloseWithEmittingDataFiles() throws Exception {
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
       testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), 2);
@@ -221,13 +228,15 @@ public void testBoundedStreamCloseWithEmittingDataFiles() throws Exception {
       ((BoundedOneInput) testHarness.getOneInputOperator()).endInput();
 
       int expectedDataFiles = partitioned ? 2 : 1;
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
       ((BoundedOneInput) testHarness.getOneInputOperator()).endInput();
 
-      result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       // Datafiles should not be sent again
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -236,7 +245,7 @@ public void testBoundedStreamCloseWithEmittingDataFiles() throws Exception {
 
   @TestTemplate
   public void testBoundedStreamTriggeredEndInputBeforeTriggeringCheckpoint() throws Exception {
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
       testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), 2);
@@ -244,13 +253,15 @@ public void testBoundedStreamTriggeredEndInputBeforeTriggeringCheckpoint() throw
       testHarness.endInput();
 
       int expectedDataFiles = partitioned ? 2 : 1;
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(expectedDataFiles);
 
       testHarness.prepareSnapshotPreBarrier(1L);
 
-      result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       // It should be ensured that after endInput is triggered, when prepareSnapshotPreBarrier
       // is triggered, write should only send WriteResult once
@@ -275,7 +286,7 @@ public void testTableWithTargetFileSize() throws Exception {
       }
     }
 
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter()) {
       for (RowData row : rows) {
         testHarness.processElement(row, 1);
@@ -283,7 +294,8 @@ public void testTableWithTargetFileSize() throws Exception {
 
       // snapshot the operator.
       testHarness.prepareSnapshotPreBarrier(1);
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(8);
 
@@ -346,13 +358,14 @@ public void testPromotedFlinkDataType() throws Exception {
             record.copy(ImmutableMap.of("tinyint", 2, "smallint", 0, "int", 102)),
             record.copy(ImmutableMap.of("tinyint", 3, "smallint", 32767, "int", 103)));
 
-    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+    try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> testHarness =
         createIcebergStreamWriter(icebergTable, flinkSchema)) {
       for (RowData row : rows) {
         testHarness.processElement(row, 1);
       }
       testHarness.prepareSnapshotPreBarrier(1);
-      WriteResult result = WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+      WriteResult result =
+          WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
       assertThat(result.deleteFiles()).isEmpty();
       assertThat(result.dataFiles()).hasSize(partitioned ? 3 : 1);
 
@@ -365,12 +378,18 @@ public void testPromotedFlinkDataType() throws Exception {
     SimpleDataUtil.assertTableRecords(location, expected);
   }
 
-  private OneInputStreamOperatorTestHarness<RowData, WriteResult> createIcebergStreamWriter()
+  private static List<WriteResult> getWriteResults(List<FlinkWriteResult> flinkWriteResults) {
+    return flinkWriteResults.stream()
+        .map(FlinkWriteResult::writeResult)
+        .collect(Collectors.toList());
+  }
+
+  private OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> createIcebergStreamWriter()
       throws Exception {
     return createIcebergStreamWriter(table, SimpleDataUtil.FLINK_SCHEMA);
   }
 
-  private OneInputStreamOperatorTestHarness<RowData, WriteResult> createIcebergStreamWriter(
+  private OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> createIcebergStreamWriter(
       Table icebergTable, TableSchema flinkSchema) throws Exception {
     RowType flinkRowType = FlinkSink.toFlinkRowType(icebergTable.schema(), flinkSchema);
     FlinkWriteConf flinkWriteConfig =
@@ -379,7 +398,7 @@ private OneInputStreamOperatorTestHarness<RowData, WriteResult> createIcebergStr
 
     IcebergStreamWriter<RowData> streamWriter =
         FlinkSink.createStreamWriter(() -> icebergTable, flinkWriteConfig, flinkRowType, null);
-    OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+    OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
         new OneInputStreamOperatorTestHarness<>(streamWriter, 1, 1, 0);
 
     harness.setup();