{"record":{"id":"bbf4ec592f7a4c63","repo":"apache/seatunnel","slug":"schema-change-after-checkpoint-is-already-complete","errorCode":null,"errorMessage":"schema-change-after checkpoint is already completed, job id: %s, pipeline id: %s, checkpoint id: %s.","messagePattern":"schema-change-after checkpoint is already completed, job id: (.+?), pipeline id: (.+?), checkpoint id: (.+?)\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java","lineNumber":1568,"sourceCode":"    }\n\n    protected void completeSchemaChangeAfterCheckpoint(CompletedCheckpoint checkpoint) {\n        if (schemaChanging.compareAndSet(true, false)) {\n            LOG.info(\n                    \"completed schema-change-after checkpoint, job id: {}, pipeline id: {}, \"\n                            + \"checkpoint id: {}.\",\n                    jobId,\n                    pipelineId,\n                    checkpoint.getCheckpointId());\n            LOG.info(\n                    \"recover trigger general-checkpoint, job id: {}, pipeline id: {}, \"\n                            + \"checkpoint id: {}.\",\n                    jobId,\n                    pipelineId,\n                    checkpoint.getCheckpointId());\n            scheduleTriggerPendingCheckpoint(coordinatorConfig.getCheckpointInterval());\n        } else {\n            throw new IllegalStateException(\n                    String.format(\n                            \"schema-change-after checkpoint is already completed, \"\n                                    + \"job id: %s, pipeline id: %s, checkpoint id: %s.\",\n                            jobId, pipelineId, checkpoint.getCheckpointId()));\n        }\n    }\n\n    public String getCheckpointStateImapKey() {\n        return checkpointStateImapKey;\n    }\n\n    public String getReadyToCloseImapKey() {\n        return readyToCloseImapKey;\n    }\n\n    /** Only for test */\n    @VisibleForTesting\n    public PendingCheckpoint getSavepointPendingCheckpoint() {","sourceCodeStart":1550,"sourceCodeEnd":1586,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java#L1550-L1586","documentation":"Thrown when the coordinator is asked to trigger a schema-change-after checkpoint, but that checkpoint has already completed — it is tracked as completed, so triggering another one is an invalid state transition. IllegalStateException signals a caller sequencing bug rather than a user config problem.","triggerScenarios":"Requesting/advancing a schema-change-after checkpoint for a (jobId, pipelineId, checkpointId) whose state is already COMPLETED — e.g., a duplicate schema-change trigger, a retried notification after the checkpoint finished, or concurrent schema-change requests racing.","commonSituations":"Schema evolution events delivered twice (retries from the catalog/source); a schema change arriving right after the schema-change checkpoint finished; concurrent connectors each triggering the same schema-change checkpoint.","solutions":["Check checkpoint state before triggering: skip if the schema-change-after checkpoint is already completed (the normal path already reschedules the interval checkpoint)","Deduplicate schema-change trigger notifications (idempotency by checkpointId) at the caller","Serialize schema-change requests per pipeline so concurrent triggers can't race past the completed check","If the error comes from a retry path, treat 'already completed' as success and continue rather than failing the job"],"exampleFix":"// before\ncoordinator.triggerSchemaChangeCheckpoint(checkpointId); // may already be completed\n\n// after\nif (!coordinator.isSchemaChangeCheckpointCompleted(checkpointId)) {\n    coordinator.triggerSchemaChangeCheckpoint(checkpointId);\n} else {\n    LOG.info(\"Schema-change checkpoint {} already completed, skipping\", checkpointId);\n}","handlingStrategy":"try-catch","validationCode":"if (isSchemaChangeCheckpointCompleted(jobId, pipelineId, checkpointId)) {\n    return; // already done, nothing to trigger\n}","typeGuard":null,"tryCatchPattern":"try {\n    triggerSchemaChangeCheckpoint(checkpointId);\n} catch (IllegalStateException e) {\n    if (e.getMessage().contains(\"already completed\")) {\n        LOG.info(\"Schema-change checkpoint already done; treating as success\");\n    } else { throw e; }\n}","preventionTips":["Make schema-change triggers idempotent per checkpointId","Serialize schema-change requests per pipeline","Ignore duplicate trigger notifications from sources/catalogs"],"tags":["zeta-engine","checkpoint","schema-change","invalid-state","race-condition"],"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-14T05:17:10.506Z"}