{"record":{"id":"c2d931ab179d0857","repo":"apache/iceberg","slug":"table-already-contains-staged-changes","errorCode":null,"errorMessage":"Table already contains staged changes.","messagePattern":"Table already contains staged changes\\.","errorType":"validation","errorClass":"MaxCommittedCheckpointMismatchException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java","lineNumber":358,"sourceCode":"\n  private static class MaxCommittedCheckpointIdValidator implements SnapshotAncestryValidator {\n    private final long stagedCheckpointId;\n    private final String flinkJobId;\n    private final String flinkOperatorId;\n\n    private MaxCommittedCheckpointIdValidator(\n        long stagedCheckpointId, String flinkJobId, String flinkOperatorId) {\n      this.stagedCheckpointId = stagedCheckpointId;\n      this.flinkJobId = flinkJobId;\n      this.flinkOperatorId = flinkOperatorId;\n    }\n\n    @Override\n    public boolean validate(Iterable<Snapshot> baseSnapshots) {\n      long maxCommittedCheckpointId =\n          getMaxCommittedCheckpointId(baseSnapshots, flinkJobId, flinkOperatorId);\n      if (maxCommittedCheckpointId >= stagedCheckpointId) {\n        throw new MaxCommittedCheckpointMismatchException();\n      }\n\n      return true;\n    }\n  }\n\n  @VisibleForTesting\n  void commitOperation(\n      Table table,\n      String branch,\n      SnapshotUpdate<?> operation,\n      CommitSummary summary,\n      String description,\n      String newFlinkJobId,\n      String operatorId,\n      long checkpointId) {\n\n    LOG.info(","sourceCodeStart":340,"sourceCodeEnd":376,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java#L340-L376","documentation":"The committer's validation hook checks that the checkpoint being committed is not older than the maximum checkpoint already committed for this flinkJobId/operatorId found in the base snapshots. If maxCommittedCheckpointId >= stagedCheckpointId, the staged changes are stale (already committed or superseded), so it throws MaxCommittedCheckpointMismatchException with this message rather than double-committing. The given SOURCE shows the equivalent validate() path that guards this invariant.","triggerScenarios":"Committing a DynamicCommittable whose stagedCheckpointId is <= the highest checkpoint id already recorded in table snapshots for the same flink job/operator, e.g. on retry after a commit actually succeeded but the response was lost, or when restoring from a savepoint older than the table state.","commonSituations":"Job restarted from a savepoint while the table already contains commits from a later checkpoint; Flink's checkpoint retry semantics racing with a slow but successful Iceberg commit; running two jobs with the same flinkJobId against one table.","solutions":["Treat as benign if the data is already committed: the commit is idempotent; verify the latest snapshot and skip/retry the checkpoint","Restore from a savepoint/checkpoint that matches the table's committed state","Ensure flinkJobId is unique per deployment so checkpoint bookkeeping isn't shared between jobs"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Before committing, check the table's latest committed checkpoint for this job:\nlong maxCommitted = getMaxCommittedCheckpointId(table.snapshots(), flinkJobId, operatorId);\nif (maxCommitted >= stagedCheckpointId) { /* skip: already committed */ }","typeGuard":null,"tryCatchPattern":"try {\n  committer.commit(committables);\n} catch (MaxCommittedCheckpointMismatchException e) {\n  // data already committed; verify latest snapshot and continue with next checkpoint\n}","preventionTips":["Restore jobs from savepoints consistent with the table's committed state","Ensure each deployment uses a unique flinkJobId","Rely on the sink's idempotent checkpoint bookkeeping rather than manual commits"],"tags":["flink","commit","checkpoint","idempotency"],"backgroundTag":"invalid-state-transition","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}