Flink: IcebergSink silently drops commits after stateless restart
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 16h
- Merged PRs (30d)
- 129
Description
### 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()`.
Contributor guide
Research direction
Start in org.apache.iceberg.flink.sink, reading SinkUtil.getMaxCommittedCheckpointId and the IcebergSink commit path; compare its lookup with the isRestored() guard in the legacy FlinkFilesCommitter/FlinkSink path. Done means a stateless restart permits subsequent checkpoints to create new Iceberg snapshots without changing restored-state behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 76/100