{"record":{"id":"6b27c889ed2c894c","repo":"apache/seatunnel","slug":"the-state-must-be-end-state-from-executionservice","errorCode":null,"errorMessage":"The state must be end state from ExecutionService, can not be ${executionState}","messagePattern":"The state must be end state from ExecutionService, can not be (.+?)","errorType":"exception","errorClass":"SeaTunnelEngineException","httpStatus":null,"severity":"error","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java","lineNumber":535,"sourceCode":"                                \"Set %s state %s to Imap failed, skip.\",\n                                getTaskFullName(), ExecutionState.CREATED));\n            }\n            this.currExecutionState = ExecutionState.CREATED;\n            log.info(String.format(\"%s turn to state %s.\", taskFullName, ExecutionState.CREATED));\n        }\n    }\n\n    public void reset() {\n        resetExecutionState();\n    }\n\n    public String getTaskFullName() {\n        return taskFullName;\n    }\n\n    public void updateStateByExecutionService(TaskExecutionState taskExecutionState) {\n        if (!taskExecutionState.getExecutionState().isEndState()) {\n            throw new SeaTunnelEngineException(\n                    String.format(\n                            \"The state must be end state from ExecutionService, can not be %s\",\n                            taskExecutionState.getExecutionState()));\n        }\n        errorByPhysicalVertex.compareAndSet(null, taskExecutionState.getThrowableMsg());\n        updateTaskState(taskExecutionState.getExecutionState());\n    }\n\n    public synchronized void forceStop() {\n        ExecutionState executionState = getExecutionState();\n        if (executionState == null || executionState.isEndState()) {\n            return;\n        }\n        noticeTaskExecutionServiceCancel();\n        if (!taskFuture.isDone()) {\n            updateTaskState(ExecutionState.CANCELED);\n        }\n    }","sourceCodeStart":517,"sourceCodeEnd":553,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java#L517-L553","documentation":"updateStateByExecutionService is the callback for ExecutionService task-completion notifications, which are only expected to deliver end states (FINISHED, FAILED, CANCELED). Receiving a non-end state means the internal contract between ExecutionService and PhysicalVertex is broken, so a SeaTunnelEngineException is thrown immediately.","triggerScenarios":"ExecutionService invokes updateStateByExecutionService with a TaskExecutionState whose execution state is RUNNING, CREATED, FAILING, etc.; an engine bug or version mismatch delivering intermediate states through this channel.","commonSituations":"Custom engine modifications that forward intermediate task states to the plan; mixed engine jar versions after an upgrade; bugs in task lifecycle reporting during failover.","solutions":["Verify all engine modules run the same SeaTunnel version (no mixed jars)","Ensure ExecutionService only notifies the vertex of terminal task states","Check for custom patches/modified ExecutionService code and revert to upstream","Capture logs of the offending TaskExecutionState and report the engine bug"],"exampleFix":"// before\nvertex.updateStateByExecutionService(new TaskExecutionState(loc, ExecutionState.RUNNING, null));\n// after\nif (state.getExecutionState().isEndState()) {\n    vertex.updateStateByExecutionService(state);\n}","handlingStrategy":"try-catch","validationCode":"if (!taskExecutionState.getExecutionState().isEndState()) { return; // do not forward to updateStateByExecutionService }","typeGuard":"boolean isTerminalTaskState(TaskExecutionState s) { return s.getExecutionState().isEndState(); }","tryCatchPattern":"try { vertex.updateStateByExecutionState(tes); } catch (SeaTunnelEngineException e) { if (e.getMessage().contains(\"must be end state from ExecutionService\")) { log.error(\"contract violation in task notification\", e); } else { throw e; } }","preventionTips":["Only route terminal task states to the vertex","Avoid mixed-version engine jars","Test custom ExecutionService changes against the vertex contract"],"tags":["zeta-engine","task-state","contract-violation","state-machine"],"backgroundTag":"internal-invariant-violation","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"}