{"record":{"id":"148a9d1175a4bf40","repo":"apache/seatunnel","slug":"schema-change-checkpoint-s-s-and-phase-s-is-n","errorCode":null,"errorMessage":"schema-change checkpoint[%s,%s] and phase[%s] is not matched","messagePattern":"schema-change checkpoint\\[(.+?),(.+?)\\] and phase\\[(.+?)\\] is not matched","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java","lineNumber":513,"sourceCode":"        }\n\n        log.debug(\n                \"trigger barrier [{}] finished, cost: {}ms. taskLocation: [{}]\",\n                barrier.getId(),\n                System.currentTimeMillis() - startTime,\n                currentTaskLocation);\n\n        CheckpointType checkpointType = ((CheckpointBarrier) barrier).getCheckpointType();\n        if (checkpointType.isSchemaChangeCheckpoint()) {\n            if (schemaChanging()) {\n                if (checkpointType.isSchemaChangeBeforeCheckpoint()\n                        && schemaChangePhase.get().isBeforePhase()) {\n                    schemaChangePhase.get().setCheckpointId(barrier.getId());\n                } else if (checkpointType.isSchemaChangeAfterCheckpoint()\n                        && schemaChangePhase.get().isAfterPhase()) {\n                    schemaChangePhase.get().setCheckpointId(barrier.getId());\n                } else {\n                    throw new IllegalStateException(\n                            String.format(\n                                    \"schema-change checkpoint[%s,%s] and phase[%s] is not matched\",\n                                    barrier.getId(),\n                                    checkpointType,\n                                    schemaChangePhase.get().getPhase()));\n                }\n                log.info(\n                        \"lock checkpoint[{}] waiting for complete..., phase: [{}]\",\n                        barrier.getId(),\n                        schemaChangePhase.get().getPhase());\n            } else {\n                log.debug(\n                        \"Ignore schema-change checkpoint[{}] on idle task, phase: [{}]\",\n                        barrier.getId(),\n                        checkpointType);\n            }\n        }\n    }","sourceCodeStart":495,"sourceCodeEnd":531,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java#L495-L531","documentation":"In SourceFlowLifeCycle.triggerBarrier(), when a barrier of schema-change type arrives, the engine verifies that the barrier's checkpoint type matches the currently recorded SchemaChangePhase (before-checkpoint pairs with before-phase, after with after). A mismatch means the phase recorded on the source path does not correspond to the barrier the coordinator sent, so an IllegalStateException is thrown.","triggerScenarios":"triggerBarrier() receives a checkpointType.isSchemaChangeBeforeCheckpoint() barrier while phase is after-phase (or vice versa), or any schema-change-typed barrier while schemaChangePhase is null/mismatched.","commonSituations":"Checkpoint type/phase bookkeeping desynchronization after a failed or retried schema-change checkpoint; barrier reordering after task restart or recovery from an older checkpoint id; mixed engine/connector versions during rolling upgrade.","solutions":["Restart the affected task so phase state and barrier stream resynchronize.","Check logs for an earlier failed triggerSchemaChangeBefore/AfterCheckpoint that left the wrong phase set.","Verify all nodes run the same SeaTunnel version (rolling-upgrade mixing of checkpoint semantics).","Retry the job from the last consistent checkpoint/snapshot."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// verify phase matches barrier type before trigger\nboolean matched = (checkpointType.isSchemaChangeBeforeCheckpoint() && phase.isBeforePhase()) || (checkpointType.isSchemaChangeAfterCheckpoint() && phase.isAfterPhase());","typeGuard":null,"tryCatchPattern":"try { triggerBarrier(barrier); } catch (IllegalStateException e) { if (e.getMessage().contains(\"is not matched\")) { restartTaskToResyncPhase(); } else { throw e; } }","preventionTips":["Keep engine versions uniform across the cluster","Investigate any earlier schema-change trigger failure immediately","Restart task from snapshot when phase/barrier mismatch is detected"],"tags":["checkpoint","schema-change","barrier","state-machine"],"backgroundTag":"invalid-state-transition","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"}