{"record":{"id":"ad531cfdf69eff1f","repo":"apache/seatunnel","slug":"the-splitchange-type-of-s-is-not-supported-ad531c","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-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java","lineNumber":167,"sourceCode":"                currentOffset);\n        finishedPartitions.add(tp);\n        recordsBySplits.addFinishedSplit(tp.toString());\n    }\n\n    private void markEmptySplitsAsFinished(KafkaPartitionSplitRecords recordsBySplits) {\n        // Some splits are discovered as empty when handling split additions. These splits should be\n        // added to finished splits to clean up states in split fetcher and source reader.\n        if (!emptySplits.isEmpty()) {\n            recordsBySplits.finishedSplits.addAll(emptySplits);\n            emptySplits.clear();\n        }\n    }\n\n    @Override\n    public void handleSplitsChanges(SplitsChange<KafkaSourceSplit> splitsChange) {\n        // Get all the partition assignments and stopping offsets.\n        if (!(splitsChange instanceof SplitsAddition)) {\n            throw new UnsupportedOperationException(\n                    String.format(\n                            \"The SplitChange type of %s is not supported.\",\n                            splitsChange.getClass()));\n        }\n\n        // Assignment.\n        List<TopicPartition> newPartitionAssignments = new ArrayList<>();\n        // Starting offsets.\n        Map<TopicPartition, Long> partitionsStartingOffsets = new HashMap<>();\n        // Stopping offsets.\n        List<TopicPartition> partitionsStoppingAtLatest = new ArrayList<>();\n\n        // Parse the starting and stopping offsets.\n        splitsChange\n                .splits()\n                .forEach(\n                        s -> {\n                            newPartitionAssignments.add(s.getTopicPartition());","sourceCodeStart":149,"sourceCodeEnd":185,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java#L149-L185","documentation":"KafkaSource's KafkaPartitionSplitReader only implements incremental split assignment via SplitsAddition. Any other SplitsChange subtype (e.g. SplitsRemove) is rejected with UnsupportedOperationException because removal/rebalancing of already-assigned splits is not supported by this reader.","triggerScenarios":"Calling handleSplitsChanges on KafkaPartitionSplitReader with a SplitsChange that is not an instance of SplitsAddition<KafkaSourceSplit>, e.g. a SplitsRemove produced during dynamic split removal or a custom SourceReader implementation passing an unsupported change type.","commonSituations":"Engine or framework code performing split removals after a reader scale-down or failed assignment; custom readers wrapping Kafka source splits; framework versions where the split change protocol changed.","solutions":["Ensure split changes sent to the Kafka source reader are SplitsAddition instances only.","If split removal is needed, recreate the job or use a source implementation that supports SplitsRemove.","Check the engine/connector version pairing so the split-change protocol matches what KafkaSourceReader supports."],"exampleFix":"// before\nsplitsChanges.add(new SplitsRemove<>(removedSplits));\n// after\nsplitsChanges.add(new SplitsAddition<>(addedSplits)); // Kafka reader supports additions only","handlingStrategy":"type-guard","validationCode":"if (!(splitsChange instanceof SplitsAddition)) { throw new IllegalArgumentException(\"Kafka reader only accepts SplitsAddition, got: \" + splitsChange.getClass()); }","typeGuard":"boolean isAddition(SplitsChange<?> c) { return c instanceof SplitsAddition; }","tryCatchPattern":"try { reader.handleSplitsChanges(change); } catch (UnsupportedOperationException e) { log.error(\"Unsupported split change type\", e); }","preventionTips":["Only emit SplitsAddition to Kafka source readers.","Do not implement custom split removal for this connector."],"tags":["kafka","source-reader","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"}