{"record":{"id":"7a9b5311c4526b27","repo":"apache/seatunnel","slug":"multiple-incremental-splits-are-not-supported","errorCode":null,"errorMessage":"Multiple incremental splits are not supported","messagePattern":"Multiple incremental splits are not supported","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/BaseChangeStreamTableSourceFactory.java","lineNumber":80,"sourceCode":"    }\n\n    public abstract <T, SplitT extends SourceSplit, StateT extends Serializable>\n            TableSource<T, SplitT, StateT> restoreSource(\n                    TableSourceFactoryContext context, List<CatalogTable> restoreTableStruct);\n\n    protected <SplitT extends SourceSplit, StateT extends Serializable>\n            List<CatalogTable> getRestoreTableStruct(\n                    ChangeStreamTableSourceState<StateT, SplitT> state) {\n        List<IncrementalSplit> incrementalSplits =\n                state.getSplits().stream()\n                        .flatMap(List::stream)\n                        .filter(e -> e != null)\n                        .map(e -> SourceSplitBase.class.cast(e))\n                        .filter(e -> e.isIncrementalSplit())\n                        .map(e -> e.asIncrementalSplit())\n                        .collect(Collectors.toList());\n        if (incrementalSplits.size() > 1) {\n            throw new UnsupportedOperationException(\n                    \"Multiple incremental splits are not supported\");\n        }\n\n        if (incrementalSplits.size() == 1) {\n            IncrementalSplit incrementalSplit = incrementalSplits.get(0);\n            if (incrementalSplit.getCheckpointTables() != null) {\n                List<CatalogTable> checkpointTableStruct = incrementalSplit.getCheckpointTables();\n                log.info(\"Restore source using checkpoint tables: {}\", checkpointTableStruct);\n                return checkpointTableStruct;\n            }\n            if (incrementalSplit.getCheckpointDataType() != null) {\n                // TODO: Waiting for remove of compatible logic\n                List<CatalogTable> checkpointDataTypeStruct =\n                        CatalogTableUtil.convertDataTypeToCatalogTables(\n                                incrementalSplit.getCheckpointDataType(), \"default.default\");\n                log.info(\"Restore source using checkpoint tables: {}\", checkpointDataTypeStruct);\n                return checkpointDataTypeStruct;\n            }","sourceCodeStart":62,"sourceCodeEnd":98,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/BaseChangeStreamTableSourceFactory.java#L62-L98","documentation":"When restoring a CDC source from checkpoint, BaseChangeStreamTableSourceFactory.getRestoreTableStruct collects restored splits and filters for incremental splits. Incremental (streaming) phase must have at most one incremental split; more than one indicates corrupted or duplicated checkpoint state, so UnsupportedOperationException is thrown.","triggerScenarios":"Calling restoreSource (which calls getRestoreTableStruct) with a restored SourceSplitBase collection where more than one split reports isIncrementalSplit() == true.","commonSituations":"Restoring a checkpoint/savepoint written by a different SeaTunnel or connector version with an incompatible split layout, manually edited/copied checkpoint data, or a custom connector that incorrectly emits multiple incremental splits.","solutions":["Delete the failed checkpoint state and restart the job from scratch (or from a consistent snapshot savepoint)","Ensure the same SeaTunnel and connector-cdc versions are used to save and restore the checkpoint","Do not manually merge or duplicate checkpoint split files; restore from an untouched savepoint","If developing a custom CDC connector, ensure only one incremental split is ever produced per source"],"exampleFix":"// before\n// restore with checkpoint from older incompatible connector version -> multiple incremental splits\nbin/seatunnel.sh --config job.conf -r old-checkpoint-id\n// after\n// upgrade both nodes, then restore from a savepoint created with the same version\nbin/seatunnel.sh --config job.conf -r consistent-savepoint-id","handlingStrategy":"fallback","validationCode":"// before restoring, verify only one incremental split exists in state\nlong n = restoredSplits.stream().filter(SourceSplitBase::isIncrementalSplit).count();\nif (n > 1) throw new IllegalStateException(\"Corrupt checkpoint: \" + n + \" incremental splits\");","typeGuard":null,"tryCatchPattern":"try {\n    factory.restoreSource(...);\n} catch (UnsupportedOperationException e) {\n    // fall back to a fresh (non-restored) submission\n}","preventionTips":["Save and restore checkpoints with identical engine and connector versions","Never hand-edit or merge checkpoint split files","Restore only from untouched savepoints"],"tags":["cdc","checkpoint","restore","unsupported-operation"],"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"}