{"record":{"id":"1119938d6bdc290a","repo":"apache/iceberg","slug":"table-already-contains-staged-changes-111993","errorCode":null,"errorMessage":"Table already contains staged changes.","messagePattern":"Table already contains staged changes\\.","errorType":"validation","errorClass":"MaxCommittedCheckpointMismatchException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/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/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java#L340-L376","documentation":"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.').","triggerScenarios":"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.","commonSituations":"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.","solutions":["Restore from the latest completed checkpoint/savepoint so staged checkpoint ids advance past already-committed ones.","Ensure each job uses a unique flinkJobId/operatorId (check job_id overwrite configs).","Verify the table's snapshots — if the data was already committed, discard the stale committables instead of retrying."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"long committed = getMaxCommittedCheckpointId(table.snapshots(), flinkJobId, flinkOperatorId); if (committed >= pendingCheckpointId) { /* already committed — skip */ }","typeGuard":null,"tryCatchPattern":"try { committer.commit(committables); } catch (MaxCommittedCheckpointMismatchException e) { /* data already committed; discard stale committables and continue */ }","preventionTips":["Always restore from the latest completed checkpoint","Use unique flink job ids per job writing to the same table","Never replay old committables after a successful commit"],"tags":["flink","commit","checkpoint"],"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-14T16:17:12.679Z"}