{"record":{"id":"2ea304c94875f96b","repo":"apache/iceberg","slug":"coordinator-unexpectedly-terminated-on-committer","errorCode":null,"errorMessage":"Coordinator unexpectedly terminated on committer ${taskId}","messagePattern":"Coordinator unexpectedly terminated on committer (.+?)","errorType":"exception","errorClass":"NotRunningException","httpStatus":null,"severity":"critical","filePath":"kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java","lineNumber":200,"sourceCode":"    }\n\n    // Reset offsets to last committed to avoid data loss.\n    LOG.info(\"Seeking to last committed offsets for worker {}.\", taskId);\n    KafkaUtils.seekToLastCommittedOffsets(context);\n  }\n\n  @Override\n  public void save(Collection<SinkRecord> sinkRecords) {\n    if (sinkRecords != null && !sinkRecords.isEmpty()) {\n      startWorker();\n      worker.save(sinkRecords);\n    }\n    processControlEvents();\n  }\n\n  private void processControlEvents() {\n    if (coordinatorThread != null && coordinatorThread.isTerminated()) {\n      throw new NotRunningException(\n          String.format(\"Coordinator unexpectedly terminated on committer %s\", taskId));\n    }\n    if (worker != null) {\n      worker.process();\n    }\n  }\n\n  private void startWorker() {\n    if (null == this.worker) {\n      LOG.info(\"Starting commit worker {}\", taskId);\n      SinkWriter sinkWriter = new SinkWriter(catalog, config);\n      worker = new Worker(config, clientFactory, sinkWriter, context);\n      worker.start();\n    }\n  }\n\n  private void startCoordinator() {\n    if (null == this.coordinatorThread) {","sourceCodeStart":182,"sourceCodeEnd":218,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java#L182-L218","documentation":"NotRunningException thrown by CommitterImpl.processControlEvents when the background coordinator thread has terminated but the committer is still being asked to process control events. The coordinator is the actor that performs commits; once it dies the sink task cannot make progress and must fail so Connect can restart it.","triggerScenarios":"An unhandled exception killed the coordinator thread (e.g. a commit failed terminally) and the committer's save()/processControlEvents() path subsequently runs; the thread's isTerminated() check fires before any event processing.","commonSituations":"Coordinator crashed earlier during a commit due to a catalog auth failure, schema mismatch, or timeout; Kafka Connect keeps the task alive until the committer notices the dead coordinator at the next control-event poll.","solutions":["Inspect the coordinator thread's earlier stack trace in worker logs to find the root-cause exception that terminated it.","Fix the underlying commit failure (catalog credentials, network access to metastore/REST catalog, schema compatibility).","Restart the connector task — the sink requires a live coordinator and cannot recover in-place.","Verify coordinator thread timeouts and the exec.awaitTermination settings are adequate for your commit latency."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  committer.save(request);\n} catch (NotRunningException e) {\n  LOG.error(\"Coordinator dead; failing sink task for restart\", e);\n  throw e; // must propagate so Connect restarts the task\n}","preventionTips":["Monitor coordinator thread liveness and earlier root-cause exceptions in logs","Harden commit dependencies: catalog credentials, network, and schema compatibility","Alert on NotRunningException — it always means an earlier coordinator crash"],"tags":["kafka-connect","coordinator","thread-crash","not-running"],"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"}