apache/iceberg · error · MaxCommittedCheckpointMismatchException

Table already contains staged changes.

Error message

Table already contains staged changes.

What it means

The dynamic sink's committer validation checks the max committed checkpoint id for this Flink job/operator against the staged checkpoint id. If a snapshot already recorded a committed checkpoint id greater than or equal to the staged one, the staged changes are stale duplicates and the commit is rejected by throwing MaxCommittedCheckpointMismatchException (whose message is 'Table already contains staged changes.').

Source

Thrown at flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:358

  private static class MaxCommittedCheckpointIdValidator implements SnapshotAncestryValidator {
    private final long stagedCheckpointId;
    private final String flinkJobId;
    private final String flinkOperatorId;

    private MaxCommittedCheckpointIdValidator(
        long stagedCheckpointId, String flinkJobId, String flinkOperatorId) {
      this.stagedCheckpointId = stagedCheckpointId;
      this.flinkJobId = flinkJobId;
      this.flinkOperatorId = flinkOperatorId;
    }

    @Override
    public boolean validate(Iterable<Snapshot> baseSnapshots) {
      long maxCommittedCheckpointId =
          getMaxCommittedCheckpointId(baseSnapshots, flinkJobId, flinkOperatorId);
      if (maxCommittedCheckpointId >= stagedCheckpointId) {
        throw new MaxCommittedCheckpointMismatchException();
      }

      return true;
    }
  }

  @VisibleForTesting
  void commitOperation(
      Table table,
      String branch,
      SnapshotUpdate<?> operation,
      CommitSummary summary,
      String description,
      String newFlinkJobId,
      String operatorId,
      long checkpointId) {

    LOG.info(

View on GitHub (pinned to 86d9c8fc54)

Solutions

1. Restore from the latest completed checkpoint/savepoint so staged checkpoint ids advance past already-committed ones.
2. Ensure each job uses a unique flinkJobId/operatorId (check job_id overwrite configs).
3. Verify the table's snapshots — if the data was already committed, discard the stale committables instead of retrying.

long committed = getMaxCommittedCheckpointId(table.snapshots(), flinkJobId, flinkOperatorId); if (committed >= pendingCheckpointId) { /* already committed — skip */ }

try { committer.commit(committables); } catch (MaxCommittedCheckpointMismatchException e) { /* data already committed; discard stale committables and continue */ }

When it happens

Trigger: Committing a DynamicCommitter's staged snapshot when getMaxCommittedCheckpointId(baseSnapshots, flinkJobId, flinkOperatorId) >= stagedCheckpointId — typically after restoring from an old checkpoint/savepoint and re-committing already-committed data.

Common situations: Job restarted from an older checkpoint and re-attempting a commit that already landed; running two jobs with the same flinkJobId against one table; replaying committables.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Table already contains staged changes.
• Unrecognized version or corrupt state: <version>
• Unknown serialize version: ${version}
• Unrecognized version or corrupt state: ${version}
• Failed to close equality delta writer

AI-assisted analysis of apache/iceberg@86d9c8fc54 (2026-09-12). Data as JSON: /api/errors/1119938d6bdc290a. Report an issue: GitHub.