{"record":{"id":"3a6a355015072fba","repo":"apache/iceberg","slug":"for-table-with-at-exception-closin","errorCode":null,"errorMessage":"[For table {} with {}[{}] at {}]: Exception closing commit service","messagePattern":"\\[For table (.+?) with (.+?)\\[(.+?)\\] at (.+?)\\]: Exception closing commit service","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java","lineNumber":145,"sourceCode":"      errorCounter.inc();\n    }\n  }\n\n  @Override\n  public void processWatermark(Watermark mark) throws Exception {\n    try {\n      if (commitService != null) {\n        commitService.close();\n      }\n\n      LOG.info(\n          DataFileRewritePlanner.MESSAGE_PREFIX + \"Successfully completed data file compaction\",\n          tableName,\n          taskName,\n          taskIndex,\n          mark.getTimestamp());\n    } catch (Exception e) {\n      LOG.warn(\n          DataFileRewritePlanner.MESSAGE_PREFIX + \"Exception closing commit service\",\n          tableName,\n          taskName,\n          taskIndex,\n          mark.getTimestamp(),\n          e);\n      output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));\n      errorCounter.inc();\n    }\n\n    // Cleanup\n    this.commitService = null;\n\n    super.processWatermark(mark);\n  }\n\n  @Override\n  public void close() throws IOException {","sourceCodeStart":127,"sourceCodeEnd":163,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java#L127-L163","documentation":"DataFileRewriteCommitter.processWatermark() closes the CommitService, which performs the actual Iceberg snapshot commit of all offered rewrite file groups. If commitService.close() throws (commit failure), it is logged with the maintenance MESSAGE_PREFIX and sent to the TaskResultAggregator error stream; the commit service reference is cleared so the next trigger starts fresh.","triggerScenarios":"Closing the CommitService at the watermark when the underlying commitFileGroups fails: CommitStateUnknownException from the catalog (commit may or may not have applied), catalog write rejection, branch snapshot conflict because the starting snapshot changed, or storage errors writing metadata.","commonSituations":"Two jobs rewriting the same table concurrently causing commit conflicts; REST/Hive catalog returning 5xx or timeouts during commit leaving state unknown; schema/partition spec changed between plan and commit; branch fast-forwarded or reset by another writer.","solutions":["Read the chained exception: for CommitStateUnknownException, check the table's snapshots to determine whether the commit landed before retrying","Re-run maintenance on the next trigger — the operator resets commitService and the rewrite is re-planned","Serialize table writers: ensure no other job (Spark rewrite procedure, other maintenance job) commits to the same branch during the window","Check catalog health/quota (Hive metastore, REST catalog) if commits fail repeatedly","Refresh stale table metadata so commit validations see the current branch state"],"exampleFix":"// before: assuming the commit failed and blindly re-running produces duplicate rewrites\nboolean committed = false;\n\n// after: resolve CommitStateUnknown by inspecting snapshots before retry\ntry {\n  commitService.close();\n} catch (Exception e) {\n  if (e.getCause() instanceof CommitStateUnknownException) {\n    // check table.snapshots() for the rewritten files to decide retry vs skip\n    resolveCommitState(table, e);\n  } else {\n    throw e;\n  }\n}","handlingStrategy":"try-catch","validationCode":"// before scheduling, confirm only one writer commits to the target branch\nif (!lockFactory.createLock().tryLock()) {\n  LOG.warn(\"Another maintenance run is active; skipping\");\n} else {\n  Table table = tableLoader.loadTable();\n  Snapshot head = table.snapshot(branch); // record head to detect concurrent commits later\n}","typeGuard":null,"tryCatchPattern":"// handle CommitStateUnknown explicitly when consuming the error side-output\nif (e.getCause() instanceof CommitStateUnknownException) {\n  // inspect table.snapshots() to determine if the commit landed; only then retry\n  resolveCommitStateAndAlert(table, e);\n} else {\n  alertOps(\"Commit failed at watermark: \" + e.getMessage());\n}","preventionTips":["Use the TriggerLockFactory (e.g. ZkLockFactory) so only one maintenance job runs per table","Avoid concurrent Spark RewriteDataFiles procedures on the same branch during Flink maintenance","Keep the catalog (REST/Hive) healthy and sized for commit traffic; monitor 5xx/timeouts","Do not expire snapshots or reset branches while a maintenance cycle is in flight"],"tags":["flink","commit","iceberg","maintenance"],"backgroundTag":"commit-state-unknown","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"}