Data duplication may occur when flink checkpoint fails

[klion26]

Apache Iceberg version

1.4.3

Query engine

Flink

Please describe the bug 🐞

Currently, the Flink sink connector logic is

1. The writer receives checkpoint barrier and flush data and sends WriteResult to the committer operator
2. The committer operator finalizes all WriteResult received from current checkpoint interval
3. commit to the table when the checkpoint is completed.

if we fail in step 2 before the finalize process[1], then data written in checkpoint i and checkpoint i+1 will squashed into one commit, this will result in the "loss" of some eq-delete data.

for example,
update for primary key A in checkpoint i this will translate to: eq-delete A + insert A,
update for primary key A in checkpoint i + 1, this will translate to eq-delete A + insert A

after the checkpoint i fails, we'll commit to the table with eq-delete A + insert A + eq-delete A + insert A this second eq-delete A will be "lost" when we read from the table.

[1]

iceberg/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java

Line 215 in 9aa354d

dataFilesPerCheckpoint.put(checkpointId, writeToManifest(checkpointId));

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[pvary]

@klion26: Could you please check #10526? Is this a duplicate of that issue?
Thanks,
Peter

[klion26]

@pvary thanks for the information, yes, this is a duplicate of #10526 , will close this issue.