Flink: IcebergSink silently drops commits after stateless restart

[Beutlin]

Apache Iceberg version

1.11.0 (iceberg-flink-runtime-2.0)

Query engine

Flink

Please describe the bug

After a stateless restart (no checkpoint/savepoint restored, isRestored() == false) of a job using the modern IcebergSink (not legacy FlinkSink), the job keeps consuming/processing records and checkpoints succeed, but no new Iceberg snapshots are ever committed again. Data/manifest files keep being written to storage, but snapshot history stays frozen. No errors or warnings — looks like a broken deserializer, except files are physically written, just never committed.

Root cause

During commit, SinkUtil.getMaxCommittedCheckpointId(table, jobId, operatorId, branch) walks the snapshot chain backwards for a snapshot with a matching (job-id, operator-id), and returns its flink.max-committed-checkpoint-id. Since job-id/operator-id stay stable across a stateless restart, it finds a snapshot from the previous run and returns its old, high checkpoint id (e.g. 13304). The new run's checkpoint counter restarts at 1, so every new commit's checkpoint id is lower than the "already committed" value and gets silently discarded — indefinitely, until the counter organically exceeds the old value.

The legacy FlinkFilesCommitter/FlinkSink guards this exact lookup with isRestored() and skips it on a stateless start. IcebergSink's commit path appears to run the lookup unconditionally.

How to reproduce

1. Run a job with IcebergSink.forRowData()...uidSuffix(...), checkpointing enabled, long enough to accumulate several commits.
2. Restart the job without restoring state (fresh stateless start, same job-id/operator-id — e.g. same deployment restarted from scratch).
3. Consumer lag stays 0, checkpoints complete, but no new snapshots appear.

Environment note

Observed on Amazon Managed Service for Apache Flink (MSF), which triggered the stateless restart. Root cause is in Iceberg's own connector code (org.apache.iceberg.flink.sink), independent of runtime — should reproduce on any Flink cluster.

Workaround

Reverting to the deprecated FlinkSink API (uidPrefix instead of uidSuffix) avoids the issue, since its getMaxCommittedCheckpointId call is guarded by isRestored().

[pvary]

For a stateful, exactly-once sink, it is questionable what should be the correct behaviour if the source is restored to a previous state, but the target table itself is not restored.

Should you have restored the table to the original state as well? If the commits are removed from the table, then the commits should happen.

Another possible solution is to update the uidSuffix.

[Beutlin]

Thanks for the quick response!

To clarify the scenario — there was no deliberate restore/rollback of anything on our side. It's not that the source was restored to an earlier state while the table stayed current; rather, this was a fully stateless restart: no checkpoint or savepoint was restored at all (isRestored() == false). The Flink job simply started fresh, so its checkpoint counter began again at a low number — which, from the table's perspective, looks like "the source went back in time," but nothing was actually rolled back on either side. The table itself was never touched or reset; it kept its full, untouched commit history from the previous run.

So restoring the table to match isn't really applicable here — there's no "original state" to restore to, since the previous run's commits were all correct and intended; we just don't want them to block new, unrelated commits from a job that has no relationship to that old checkpoint numbering.

Re: updating uidSuffix — that matches our understanding too, and sounds like the right-shaped workaround (rotate the operator UID on every stateless deployment so the (job-id, operator-id) lookup in SinkUtil.getMaxCommittedCheckpointId no longer matches old snapshots). We ended up reverting to the legacy FlinkSink API instead (which guards that lookup with isRestored()), but rotating uidSuffix seems like a cleaner fix if we want to stay on IcebergSink.

Given that, would it make sense for IcebergSink's commit path to also gate the getMaxCommittedCheckpointId lookup on isRestored(), the way the legacy committer does? That would make a stateless restart behave consistently regardless of whether uidSuffix happens to be reused.

[pvary]

If you restart the job without state, then why do you keep the jobId unchanged?