{"record":{"id":"f313ede1147994f4","repo":"apache/seatunnel","slug":"the-checkpoint-coordinator-s-don-t-exist","errorCode":null,"errorMessage":"The checkpoint coordinator(%s) don't exist","messagePattern":"The checkpoint coordinator\\((.+?)\\) don't exist","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java","lineNumber":245,"sourceCode":"        }\n    }\n\n    protected void handleCheckpointError(int pipelineId, boolean neverRestore) {\n        jobMaster.handleCheckpointError(pipelineId, neverRestore);\n    }\n\n    private CheckpointCoordinator getCheckpointCoordinator(TaskLocation taskLocation) {\n        return getCheckpointCoordinator(taskLocation.getPipelineId());\n    }\n\n    public void reportCheckpointErrorFromTask(TaskLocation taskLocation, String errorMsg) {\n        getCheckpointCoordinator(taskLocation).reportCheckpointErrorFromTask(errorMsg);\n    }\n\n    public CheckpointCoordinator getCheckpointCoordinator(int pipelineId) {\n        CheckpointCoordinator coordinator = coordinatorMap.get(pipelineId);\n        if (coordinator == null) {\n            throw new RuntimeException(\n                    String.format(\"The checkpoint coordinator(%s) don't exist\", pipelineId));\n        }\n        return coordinator;\n    }\n\n    /**\n     * Called by the {@link Task}. <br>\n     * used by Task to report the {@link SeaTunnelTaskState} of the state machine.\n     */\n    public void reportedTask(TaskReportStatusOperation reportStatusOperation) {\n        // task address may change during restore.\n        log.debug(\n                \"reported task({}) status {}\",\n                reportStatusOperation.getLocation().getTaskID(),\n                reportStatusOperation.getStatus());\n        getCheckpointCoordinator(reportStatusOperation.getLocation())\n                .reportedTask(reportStatusOperation);\n    }","sourceCodeStart":227,"sourceCodeEnd":263,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java#L227-L263","documentation":"CheckpointManager.getCheckpointCoordinator looks up the CheckpointCoordinator for a pipeline in coordinatorMap; if none is registered for that pipelineId it throws a RuntimeException stating the coordinator doesn't exist. This means checkpoint operations (report progress, errors, etc.) are being routed to a pipeline the manager is not tracking on this node.","triggerScenarios":"Calling getCheckpointCoordinator(pipelineId) (or task-based report APIs that delegate to it) with a pipelineId whose coordinator was never created, or was removed after the pipeline finished/failed; also after job cleanup while tasks still emit checkpoint messages.","commonSituations":"Task checkpoints arriving after pipeline completion (stale messages in flight); master failover where the new master hasn't restored that pipeline's coordinator; wrong pipelineId passed by custom code or a buggy connector.","solutions":["Verify the pipelineId belongs to the job and the pipeline is still running before calling","Check whether the pipeline already completed/failed — in-flight checkpoint reports after completion should be ignored, not fatal","Guard callers (e.g., reportCheckpointErrorFromTask) to catch/handle the missing-coordinator case for late messages","After master failover, ensure CheckpointManager restore completed before processing checkpoint messages","Log and inspect the jobId/pipelineId in the message to confirm it isn't from a previous job run with reused IDs"],"exampleFix":"// before\nCheckpointCoordinator c = checkpointManager.getCheckpointCoordinator(pipelineId);\nc.reportCheckpointErrorFromTask(errorMsg);\n\n// after\ntry {\n    checkpointManager.getCheckpointCoordinator(pipelineId)\n        .reportCheckpointErrorFromTask(errorMsg);\n} catch (RuntimeException e) {\n    LOG.warn(\"Checkpoint coordinator for pipeline {} gone; late task report dropped\", pipelineId);\n}","handlingStrategy":"try-catch","validationCode":"CheckpointCoordinator c = coordinatorMap.get(pipelineId);\nif (c == null) { LOG.warn(\"No coordinator for pipeline {}\", pipelineId); return; }","typeGuard":"Optional<CheckpointCoordinator> findCoordinator(int pipelineId) {\n    return Optional.ofNullable(coordinatorMap.get(pipelineId));\n}","tryCatchPattern":"try {\n    manager.getCheckpointCoordinator(pipelineId).reportCheckpointErrorFromTask(msg);\n} catch (RuntimeException e) {\n    LOG.warn(\"Coordinator for pipeline {} missing; dropping late report\", pipelineId);\n}","preventionTips":["Drop (don't fail on) checkpoint messages for finished pipelines","Complete CheckpointManager restore before processing checkpoint traffic after failover","Validate pipelineId against the current job before reporting"],"tags":["zeta-engine","checkpoint","not-found","pipeline","failover"],"backgroundTag":"resource-not-found","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}