{"record":{"id":"385a8a66a8ae22b6","repo":"apache/iceberg","slug":"coordinator-taskid-is-terminated-commit-aborte","errorCode":null,"errorMessage":"Coordinator ${taskId} is terminated, commit aborted","messagePattern":"Coordinator (.+?) is terminated, commit aborted","errorType":"exception","errorClass":"ConnectException","httpStatus":null,"severity":"error","filePath":"kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java","lineNumber":299,"sourceCode":"\n    List<DataFile> dataFiles =\n        payloads.stream()\n            .filter(payload -> payload.dataFiles() != null)\n            .flatMap(payload -> payload.dataFiles().stream())\n            .filter(dataFile -> dataFile.recordCount() > 0)\n            .filter(distinctByKey(ContentFile::location))\n            .collect(Collectors.toList());\n\n    List<DeleteFile> deleteFiles =\n        payloads.stream()\n            .filter(payload -> payload.deleteFiles() != null)\n            .flatMap(payload -> payload.deleteFiles().stream())\n            .filter(deleteFile -> deleteFile.recordCount() > 0)\n            .filter(distinctByKey(ContentFile::location))\n            .collect(Collectors.toList());\n\n    if (terminated) {\n      throw new ConnectException(\n          String.format(\"Coordinator %s is terminated, commit aborted\", taskId));\n    }\n\n    if (dataFiles.isEmpty() && deleteFiles.isEmpty()) {\n      LOG.info(\n          \"Coordinator {} found nothing to commit to table {}, skipping\", taskId, tableIdentifier);\n    } else {\n      if (deleteFiles.isEmpty()) {\n        AppendFiles appendOp =\n            table.newAppend().validateWith(offsetValidator(tableIdentifier, committedOffsets));\n        if (branch != null) {\n          appendOp.toBranch(branch);\n        }\n        appendOp.set(snapshotOffsetsProp, offsetsJson);\n        appendOp.set(COMMIT_ID_SNAPSHOT_PROP, commitState.currentCommitId().toString());\n        appendOp.set(TASK_ID_SNAPSHOT_PROP, taskId);\n        if (validThroughTs != null) {\n          appendOp.set(VALID_THROUGH_TS_SNAPSHOT_PROP, validThroughTs.toString());","sourceCodeStart":281,"sourceCodeEnd":317,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java#L281-L317","documentation":"ConnectException raised in Coordinator.commitToTable just before writing accumulated data/delete files to a table, when the coordinator's terminated flag is set. It prevents committing against a coordinator that is shutting down, which could leave a partial or duplicated commit pipeline.","triggerScenarios":"stop()/terminate() was invoked (task shutdown or rebalance) while the coordinator was still inside doCommit; the terminated flag is checked immediately before the table write, so commits racing shutdown abort.","commonSituations":"Long-running commits (many files, slow catalog) overlapping a Connect task rebalance or connector stop; operator restarts the connector while a commit is in flight.","solutions":["Rely on Kafka Connect's exactly-once redelivery: let the aborted commit be retried by the next coordinator incarnation.","Reduce commit latency (smaller commit.interval.ms overlap windows, fewer files per commit) so commits finish before shutdown.","Check for repeated occurrences: frequent termination races often indicate an operator or rebalance loop that should be fixed first.","If deliberate shutdown, no action needed — this is expected abort-on-shutdown behavior."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  coordinator.doCommit();\n} catch (ConnectException e) {\n  if (String.valueOf(e.getMessage()).contains(\"is terminated, commit aborted\")) {\n    LOG.warn(\"Commit aborted due to shutdown; will be retried by next coordinator\");\n  } else {\n    throw e;\n  }\n}","preventionTips":["Avoid restarting connectors while commits are in flight; check commit activity first","Keep commit durations short relative to shutdown timeouts","Treat this as safe-to-retry: the commit protocol redelivers on the next task"],"tags":["kafka-connect","coordinator","shutdown-race","commit"],"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"}