DynamicCommitter dedup can allow duplicate commits after Flink job restart/rescale when a prior commit succeeded on the catalog but the response was lost #16008

Apache Iceberg version

1.10.1 (latest release)

Query engine

Flink

Please describe the bug 🐞

Affected component

iceberg-flink — org.apache.iceberg.flink.sink.dynamic.DynamicCommitter / DynamicWriteResultAggregator

Affected versions

Observed on main in flink/v2.0/, v2.1/, and v1.20/ (the relevant logic is identical across all three).

Summary

DynamicCommitter deduplicates commits on the triplet (flink.job-id, flink.operator-id, flink.max-committed-checkpoint-id) stored in snapshot summaries. The dedup is unsafe when:

1. A commit succeeds on the catalog but the client never observes success (timeout, network partition, TM failure mid-commit).
2. The Flink job is subsequently restarted with a new JobID (via SavepointRestore, rescale, or high-availability failover).

On restart, DynamicWriteResultAggregator re-stamps restored committables with the current (new) JobID, so the ancestor walk in DynamicCommitter.getMaxCommittedCheckpointId cannot find the successfully-committed snapshot (which carries the old JobID). The committable is committed a second time — duplicate rows in the table.

Root cause

operator-id is stable, job-id is not

DynamicWriteResultAggregator.open() (v2.0 line 80-81):

this.flinkJobId = getContainingTask().getEnvironment().getJobID().toString();
this.operatorId = getOperatorID().toString();

• operatorId is derived deterministically from the operator UID (murmur hash) — stable across restarts when uidPrefix is set, which is the recommended usage.
• flinkJobId is the current job's JobID — fresh on every submitJob call.

Restored committables are re-stamped with the new JobID

DynamicWriteResultAggregator.prepareSnapshotPreBarrier() (v2.0 line 96-111) produces committables using the currently-captured flinkJobId. If operator state (resultsByTableKeyAndSpec) is restored from savepoint but the commit of the last checkpoint was lost, the aggregator re-emits the committable with the new JobID instead of the one under which the commit originally succeeded.

Dedup keys on job-id first

DynamicCommitter.getMaxCommittedCheckpointId (v2.0 line 184-203):

if (flinkJobId.equals(snapshotFlinkJobId)
    && (snapshotOperatorId == null || snapshotOperatorId.equals(operatorId))) {

Only snapshots whose summary matches the current flinkJobId are considered. A snapshot committed by the same operator under a prior JobID is invisible → maxCommittedCheckpointId returns INITIAL_CHECKPOINT_ID (-1L) → the committable is treated
as fresh.

The CAS-time MaxCommittedCheckpointIdValidator (v2.0 line 335-357) uses the same key and does not catch this case either.

Failure sequence (end-to-end)

1. Checkpoint N completes. Writer produces manifest; committable (jobId=J1, opId=O1, ckpt=N) is delivered to DynamicCommitter.
2. operation.commit() succeeds on the catalog (snapshot written with job-id=J1, operator-id=O1, max-committed-checkpoint-id=N) but the client never receives confirmation (timeout, connection reset, TM process death between REST ack and state persist, etc.).
3. Flink treats the checkpoint as not-yet-committed. Job fails or is rescaled via savepoint.
4. New job starts with JobID=J2. State is restored; aggregator's resultsByTableKeyAndSpec comes back with the unacked data.
5. On the next prepareSnapshotPreBarrier, aggregator emits a committable (jobId=J2, opId=O1, ckpt=N) — re-stamped with J2.
6. Committer ancestor walk looks for snapshots with job-id=J2 → none exist (the J1 snapshot is present but filtered out) → maxCommittedCheckpointId = -1.
7. MaxCommittedCheckpointIdValidator is constructed with J2 → also does not see the J1 snapshot → no mismatch raised.
8. Commit proceeds. The same data files are added again. Duplicate rows.

Minimal repro (sketch)

1. Run a DynamicIcebergSink against any catalog, with a stable uidPrefix and checkpointing enabled.
2. Produce a batch that completes checkpoint N.
3. Intercept the catalog commit() call so it returns success on the server but throws a timeout / closes the connection client-side.
4. Let Flink fail the job and resume from savepoint (or trigger an external savepoint + fresh submitJob).
5. Observe two snapshots in the table whose summaries carry flink.max-committed-checkpoint-id=N, flink.operator-id=O1, and different flink.job-id values — both containing the same data files.

Severity

Data correctness (silent duplication). No exception surfaces; the duplication is observable only by querying the table or by inspecting snapshot summaries. Expected to be rare in most deployments, but materially more likely under:

• Slow/unreliable storage backing the catalog (our case: HDFS datanode intermittency).
• Long commit latencies relative to the Flink RPC timeout.
• Frequent rescaling via savepoint+resubmit (Flink Operator autoscaler, session-cluster multiplex patterns).

Notes

• The same pattern likely exists in the non-dynamic IcebergFilesCommitter; we haven't fully verified, but the dedup structure appears analogous and worth reviewing together.

References

• iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:127-203, 329-400
• iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java:78-111
• iceberg/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:146-167

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