{"record":{"id":"82e5af8e4231542c","repo":"apache/seatunnel","slug":"unsupported-close-starting-task","errorCode":null,"errorMessage":"Unsupported close starting task","messagePattern":"Unsupported close starting task","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java","lineNumber":622,"sourceCode":"                            true,\n                            ExceptionUtil::isOperationNeedRetryException,\n                            Constant.OPERATION_RETRY_SLEEP));\n        } catch (Exception e) {\n            LOG.error(\n                    \"Failed to persist readyToCloseStartingTask to IMap after retries, key: {}.\"\n                            + \" Failing the job to avoid an unrecoverable stuck state on master failover.\",\n                    readyToCloseImapKey,\n                    e);\n            throw new RuntimeException(\n                    \"Failed to persist readyToCloseStartingTask to IMap, key: \"\n                            + readyToCloseImapKey,\n                    e);\n        }\n    }\n\n    protected void readyToCloseIdleTask(TaskLocation taskLocation) {\n        if (plan.getStartingSubtasks().contains(taskLocation)) {\n            throw new UnsupportedOperationException(\"Unsupported close starting task\");\n        }\n\n        LOG.info(\n                \"Received close idle task, task id: {}, pipeline id: {}, job id: {}, detail: {}\",\n                taskLocation.getTaskID(),\n                taskLocation.getPipelineId(),\n                taskLocation.getJobId(),\n                taskLocation);\n        synchronized (readyToCloseIdleTask) {\n            if (readyToCloseIdleTask.contains(taskLocation)\n                    || closedIdleTask.contains(taskLocation)) {\n                LOG.warn(\n                        \"task already in closed, task id: {}, pipeline id: {}, job id: {}, detail: {}\",\n                        taskLocation.getTaskID(),\n                        taskLocation.getPipelineId(),\n                        taskLocation.getJobId(),\n                        taskLocation);\n                return;","sourceCodeStart":604,"sourceCodeEnd":640,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java#L604-L640","documentation":"CheckpointCoordinator.readyToCloseIdleTask handles close requests for idle (non-starting) tasks; if the requested TaskLocation is one of the plan's starting subtasks, it throws UnsupportedOperationException because starting tasks must be closed through the dedicated readyToCloseStartingTask path, not the idle-task path. This is a protocol misuse guard, not a runtime failure.","triggerScenarios":"Calling readyToCloseIdleTask(taskLocation) where taskLocation is contained in plan.getStartingSubtasks() — i.e., a source/starting task's close notification routed through the idle-task handler.","commonSituations":"Custom checkpoint/task lifecycle code or a patched connector sending close signals for source tasks via the wrong API; internal routing bugs in the close protocol; manually replaying close messages against the coordinator.","solutions":["Route close requests for starting subtasks through readyToClose (the starting-task path) instead of readyToCloseIdleTask","Check plan.getStartingSubtasks() and branch before invoking the idle-task handler","If seen in stock SeaTunnel, capture job/pipeline/task IDs and report — it likely indicates an internal routing bug","Ensure connectors don't emit close-completion signals for source tasks through the idle notification channel"],"exampleFix":"// before\ncoordinator.readyToCloseIdleTask(taskLocation); // may be a starting task\n\n// after\nif (plan.getStartingSubtasks().contains(taskLocation)) {\n    coordinator.readyToCloseStartingTask(taskLocation);\n} else {\n    coordinator.readyToCloseIdleTask(taskLocation);\n}","handlingStrategy":"validation","validationCode":"if (plan.getStartingSubtasks().contains(taskLocation)) {\n    throw new IllegalArgumentException(\"use readyToClose for starting task \" + taskLocation);\n}","typeGuard":"boolean isStartingTask(TaskLocation loc) { return plan.getStartingSubtasks().contains(loc); }","tryCatchPattern":"try {\n    coordinator.readyToCloseIdleTask(taskLocation);\n} catch (UnsupportedOperationException e) {\n    coordinator.readyToCloseStartingTask(taskLocation);\n}","preventionTips":["Branch on plan.getStartingSubtasks() before choosing the close path","Never emit idle-close signals for source/starting tasks","Keep task close routing in one well-tested helper"],"tags":["zeta-engine","checkpoint","unsupported-operation","task-lifecycle"],"backgroundTag":"unsupported-operation","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}