You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "Beutlin (via GitHub)" <gi...@apache.org> on 2026/09/14 07:58:31 UTC
[I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Beutlin opened a new issue, #18098:
URL: https://github.com/apache/iceberg/issues/18098
### 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()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Posted by "developer-rpai (via GitHub)" <gi...@apache.org>.
developer-rpai commented on issue #18098:
URL: https://github.com/apache/iceberg/issues/18098#issuecomment-5843734026
I've opened PR #18270 with a fix for this.
**Root cause (confirmed on main):** `IcebergCommitter.commit()` (the Sink V2 committer used by `IcebergSink`) calls `SinkUtil.getMaxCommittedCheckpointId(table, jobId, operatorId, branch)` unconditionally. Since `(job-id, operator-id)` stay stable across a stateless restart, the lookup matches the previous run's snapshots and returns its high checkpoint id, so every new committable (checkpoint counter restarted at 1) is marked already-committed and never committed — silent data loss.
**Fix:** gate the lookup on whether the job was restored, mirroring the legacy `FlinkSink`/`FlinkFilesCommitter` `isRestored()` guard. `IcebergSink.createCommitter()` now passes `context.getRestoredCheckpointId().isPresent()` into `IcebergCommitter`, which skips the history lookup on a fresh start (starting from `INITIAL_CHECKPOINT_ID`, like the legacy path). Applied to all four Flink modules (v1.20/v2.1/v2.2/v2.3), plus a regression test simulating the stateless-restart scenario.
Happy to adjust the approach if maintainers prefer a different fix shape.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Posted by "Beutlin (via GitHub)" <gi...@apache.org>.
Beutlin commented on issue #18098:
URL: https://github.com/apache/iceberg/issues/18098#issuecomment-5671414528
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Posted by "Beutlin (via GitHub)" <gi...@apache.org>.
Beutlin commented on issue #18098:
URL: https://github.com/apache/iceberg/issues/18098#issuecomment-5869001078
There's already an open PR for this fix: #18123 (same approach — gate the SinkUtil.getMaxCommittedCheckpointId lookup on restored state). Might be worth avoiding duplicate review effort.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on issue #18098:
URL: https://github.com/apache/iceberg/issues/18098#issuecomment-5666097778
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on issue #18098:
URL: https://github.com/apache/iceberg/issues/18098#issuecomment-5866290811
We should fix this for the DynamicSink also
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on issue #18098:
URL: https://github.com/apache/iceberg/issues/18098#issuecomment-5865134857
@Guosmilesmile: You can use FlinkWriteOptions.UID_SUFFIX to set the uid from SQL
@Beutlin: Please fix the PR title, to describe the change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on issue #18098:
URL: https://github.com/apache/iceberg/issues/18098#issuecomment-5677644537
If you restart the job without state, then why do you keep the jobId unchanged?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Flink: IcebergSink silently drops commits after stateless restart [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on issue #18098:
URL: https://github.com/apache/iceberg/issues/18098#issuecomment-5862394054
I’d prefer to keep the Sink V2 behavior consistent with V1, since a stateless restart shouldn’t silently drop data.
1. IcebergFilesCommitter in Sink V1 already handles this case: when the job is not restored from state, maxSnapshot is set to -1. Sink V2 doesn’t have this handling, so the behavior has changed.https://github.com/apache/iceberg/blob/e4bcaa40cbb4de6431584cb2a5816cb626c34ca5/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L170-L180
2. There are cases where the job can’t be restored from state and has to be restarted directly. If the topology and HA ID stay the same, the Job ID may also remain unchanged. In that case, V2 may use the old dedup information and incorrectly treat new committables as already committed, resulting in silent data loss.
3. For SQL jobs, we can’t directly set or change the UID, so we can’t rely on changing the UID to work around this.
@pvary WDYT？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org