[Bug] IcebergCommitCallback crashes the Flink job on CommitStateUnknownException even though the committer already self-heals on the next checkpoint
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.
### Paimon version
1.4.1 (also checked: the same propagation/commit-callback behavior is present on current `master`)
### Compute Engine
Flink 2.0, Paimon table with Iceberg metadata sync enabled against a REST catalog (`metadata.iceberg.storage = rest-catalog`)
### Minimal reproduce step
1. Configure a Paimon table with Iceberg REST metadata sync enabled, written to continuously by a Flink streaming job (one Iceberg commit attempt per checkpoint).
2. Cause the Iceberg REST catalog's commit endpoint to respond with a `500`/timeout on a single commit POST. The Iceberg REST client surfaces this as `org.apache.iceberg.exceptions.CommitStateUnknownException`.
3. This exception propagates unhandled out of `IcebergRestMetadataCommitter.commitMetadataImpl` → `IcebergCommitCallback.createMetadataWithBase`/`createMetadata` → `IcebergCommitCallback.call` → `FileStoreCommitImpl.tryCommitOnce` → ... → `CommitterOperator.notifyCheckpointComplete`. Because this happens inside `notifyCheckpointComplete`, Flink treats it as fatal and restarts the whole job:
```
java.lang.RuntimeException: java.lang.RuntimeException: Fail to commit metadata to rest catalog.
at org.apache.paimon.iceberg.IcebergRestMetadataCommitter.commitMetadata(IcebergRestMetadataCommitter.java:127)
at org.apache.paimon.iceberg.IcebergCommitCallback.createMetadataWithBase(IcebergCommitCallback.java:667)
at org.apache.paimon.iceberg.IcebergCommitCallback.createMetadata(IcebergCommitCallback.java:276)
at org.apache.paimon.iceberg.IcebergCommitCallback.call(IcebergCommitCallback.java:224)
at org.apache.paimon.operation.FileStoreCommitImpl.lambda$tryCommitOnce$11(FileStoreCommitImpl.java:1048)
...
at org.apache.paimon.flink.sink.CommitterOperator.commitUpToCheckpoint(CommitterOperator.java:215)
at org.apache.paimon.flink.sink.CommitterOperator.notifyCheckpointComplete(CommitterOperator.java:192)
...
Caused by: org.apache.iceberg.exceptions.CommitStateUnknownException: Service failed: 500: timeout exceeded
Cannot determine whether the commit was successful or not, the underlying data files may or may not be needed. ...
```
Crucially, this callback runs *after* Paimon's own core table commit (`FileStoreCommitImpl.commit`) has already succeeded — the Iceberg REST sync is a best-effort side effect, not part of the table's core commit path or durability guarantee.
4. On the very next checkpoint's commit attempt, `commitMetadataImpl` reloads the table fresh from the REST catalog (`icebergTable = getTable()` → `restCatalog.loadTable(...)`, not a cached/stale reference) and calls `checkBase()`:
```java
return currentMetadata.currentSnapshot().snapshotId()
== newMetadata.currentSnapshot().snapshotId() - 1;
```
What happens here depends on whether the ambiguous commit from step 2 actually landed server-side:
- **If it landed**, the REST-side current snapshot is now exactly one behind the new local metadata's current snapshot, `checkBase()` returns `true`, and the commit proceeds normally via `updatesForCorrectBase(..., false)` — no recreate needed at all.
- **If it didn't land** (or the REST side otherwise ends up more than one snapshot behind — this is also unconditionally true whenever the local `baseIcebergMetadata` passed in is `null`, a separate trigger for the same branch), `checkBase()` returns `false`, and `updatesForIncorrectBase()` → `recreateTable()` runs: it drops and recreates the Iceberg table with a single snapshot that correctly represents the current live file set (manifests are built cumulatively, so there's no query-visible data loss).
In neither branch does whether the job crashed in step 3 change the outcome, since the check is always against live server state.
### What doesn't meet your expectations?
Given step 4, it's not obvious to us that crashing the whole Flink job in step 3 helps: if the ambiguous commit actually landed, the next checkpoint would have committed normally with no recreate at all; if it didn't land, the committer self-heals via `updatesForIncorrectBase()`/`recreateTable()` on the next attempt regardless of whether the job restarted. In both cases the crash adds a full job restart (TaskManager churn, checkpoint restore, backpressure/lag buildup across *every* table this job writes, not only the one whose commit was ambiguous) on top of an outcome that seems to resolve itself either way.
We'd be curious whether there's a reason `CommitStateUnknownException` (and similarly ambiguous/retryable REST-catalog errors) is treated as fatal here rather than caught and logged, letting the next checkpoint's commit attempt resolve things via the existing self-healing path — or whether we're missing a case where crashing is actually necessary. Interested in maintainer/community input on this before we look at a PR.
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by reading IcebergCommitCallback.call/createMetadata, IcebergRestMetadataCommitter.commitMetadataImpl, and FileStoreCommitImpl.tryCommitOnce, then reproduce the REST-catalog timeout with Flink 2.0. Trace how CommitterOperator.notifyCheckpointComplete handles the exception and add coverage for the intended next-checkpoint behavior, with completion defined by maintainer agreement and a regression test for the ambiguous commit case.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100