{"record":{"id":"826641a24b77171e","repo":"apache/seatunnel","slug":"the-splitchange-type-of-s-is-not-supported-826641","errorCode":null,"errorMessage":"The SplitChange type of %s is not supported.","messagePattern":"The SplitChange type of (.+?) is not supported\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-fluss/src/main/java/org/apache/seatunnel/connectors/seatunnel/fluss/source/FlussSourceSplitReader.java","lineNumber":211,"sourceCode":"     * reports the split in {@code finishedSplits}, yet the base only drops its state one {@code\n     * pollNext} later (in {@code finishCurrentFetch}), after the split's last record has already\n     * advanced {@code currentOffset} to the end. A checkpoint landing in that window persists a\n     * just-completed split, and restoring it hands us back {@code start == end}. Finishing it here\n     * skips a pointless subscribe + poll and avoids a misleading out-of-range warning should\n     * retention have meanwhile passed that end.\n     *\n     * <p>The {@link LogScanner#EARLIEST_OFFSET} (-2) sentinel of a fresh split floors to 0, so a\n     * fresh split is only drained when its end offset is 0 — an empty bucket, which the enumerator\n     * already filters out, leaving that a defensive fallback.\n     */\n    static boolean isDrainedAtAssignment(long startOffset, long endOffset) {\n        return isBounded(endOffset) && Math.max(startOffset, 0L) >= endOffset;\n    }\n\n    @Override\n    public void handleSplitsChanges(SplitsChange<FlussSourceSplit> splitsChanges) {\n        if (!(splitsChanges instanceof SplitsAddition)) {\n            throw new UnsupportedOperationException(\n                    String.format(\n                            \"The SplitChange type of %s is not supported.\",\n                            splitsChanges.getClass()));\n        }\n        for (FlussSourceSplit split : splitsChanges.splits()) {\n            long startOffset = split.getStartOffset();\n            long endOffset = split.getEndOffset();\n            if (isDrainedAtAssignment(startOffset, endOffset)) {\n                drainedSplits.add(split.splitId());\n                log.info(\n                        \"Split {} has nothing to read (startOffset={}, endOffset={}); marking finished without subscribing\",\n                        split.splitId(),\n                        startOffset,\n                        endOffset);\n                continue;\n            }\n            if (tableScan == null) {\n                tableScan = createTableScan();","sourceCodeStart":193,"sourceCodeEnd":229,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-fluss/src/main/java/org/apache/seatunnel/connectors/seatunnel/fluss/source/FlussSourceSplitReader.java#L193-L229","documentation":"FlussSourceSplitReader only accepts SplitsAddition changes; any other SplitsChange subtype (e.g. SplitsRemoval) is rejected with this UnsupportedOperationException. The reader cannot retract or remove already-subscribed splits.","triggerScenarios":"handleSplitsChanges(SplitsChange<FlussSourceSplit>) receives a change instance that is not SplitsAddition — typically SplitsRemoval sent during split reassignment, failover rebalancing, or by custom enumerator logic.","commonSituations":"A rollback/restore scenario where the framework replays split ownership changes; custom enumerator code issuing SplitsRemoval; older SeaTunnel runtimes emitting removal changes to release splits.","solutions":["Ensure only SplitsAddition is sent to this reader (enumerator should not reassign splits of live readers)","In restore/rebalance paths, recreate the reader instead of sending SplitsRemoval","Patch handleSplitsChanges to no-op or handle SplitsRemoval if the runtime legitimately sends it","Upgrade connector/runtime versions so split redistribution matches the push-only model"],"exampleFix":"// before (enumerator)\nenumeratorContext.assignSplit(new SplitsRemoval<>(splits, subtaskId));\n// after\n// do not send removals; splits stay with their reader until task teardown","handlingStrategy":"type-guard","validationCode":"if (!(change instanceof SplitsAddition)) {\n  throw new IllegalArgumentException(\"Only SplitsAddition is supported by FlussSourceSplitReader\");\n}","typeGuard":"boolean isAddition(SplitsChange<?> c) { return c instanceof SplitsAddition; }","tryCatchPattern":"try {\n  reader.handleSplitsChanges(change);\n} catch (UnsupportedOperationException e) {\n  LOG.warn(\"Rejected split change {} for Fluss reader\", change.getClass(), e);\n}","preventionTips":["Only send SplitsAddition to FlussSourceSplitReader","Avoid SplitsRemoval in enumerator rebalance logic for this connector","Test restore/rebalance paths to confirm no removal changes are emitted"],"tags":["unsupported-operation","split-reader","fluss"],"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"}