apache / apache/iceberg

Flink: IcebergSink silently drops commits after stateless restart

Open
#18,098 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.