{"record":{"id":"042202541b764de4","repo":"apache/seatunnel","slug":"please-invoke-deserializationschema-deserialize-by","errorCode":null,"errorMessage":"Please invoke DeserializationSchema#deserialize(byte[], Collector<SeaTunnelRow>) instead.","messagePattern":"Please invoke DeserializationSchema#deserialize\\(byte\\[\\], Collector<SeaTunnelRow>\\) instead\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java","lineNumber":96,"sourceCode":"    public CompatibleKafkaConnectDeserializationSchema(\n            @NonNull CatalogTable catalogTable,\n            boolean keySchemaEnable,\n            boolean valueSchemaEnable,\n            boolean failOnMissingField,\n            boolean ignoreParseErrors) {\n        this.catalogTable = catalogTable;\n        this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();\n        this.keySchemaEnable = keySchemaEnable;\n        this.valueSchemaEnable = valueSchemaEnable;\n        // Runtime converter\n        this.runtimeConverter =\n                new JsonToRowConverters(failOnMissingField, ignoreParseErrors)\n                        .createRowConverter(checkNotNull(seaTunnelRowType));\n    }\n\n    @Override\n    public SeaTunnelRow deserialize(byte[] message) throws IOException {\n        throw new UnsupportedOperationException(\n                \"Please invoke DeserializationSchema#deserialize(byte[], Collector<SeaTunnelRow>) instead.\");\n    }\n\n    /**\n     * Deserialize kafka consumer record\n     *\n     * @param msg\n     * @param out\n     * @throws Exception\n     */\n    public void deserialize(ConsumerRecord<byte[], byte[]> msg, Collector<SeaTunnelRow> out)\n            throws InvocationTargetException, IllegalAccessException {\n        tryInitConverter();\n        if (msg == null) {\n            return;\n        }\n        SinkRecord record = convertToSinkRecord(msg);\n        RowKind rowKind = RowKind.INSERT;","sourceCodeStart":78,"sourceCodeEnd":114,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java#L78-L114","documentation":"CompatibleKafkaConnectDeserializationSchema intentionally does not implement the single-argument deserialize(byte[]). Kafka Connect JSON records need the full ConsumerRecord (headers, key/value schemas), so only the deserialize(byte[], Collector) overload carries enough context. Calling the byte[]-only form always throws this UnsupportedOperationException.","triggerScenarios":"Calling schema.deserialize(message) directly, or plugging this schema into a connector/framework that invokes the byte[]-only DeserializationSchema API instead of the Collector overload.","commonSituations":"Custom Kafka source code calling the wrong overload in tests; a connector version that still uses the old DeserializationSchema contract; unit tests copied from plain-JSON-format examples.","solutions":["Call deserialize(byte[] message, Collector<SeaTunnelRow> out) instead, wrapping the raw bytes in a ConsumerRecord if needed.","Use SeaTunnel's Kafka connector which routes through deserialize(SourceRecord/ConsumerRecord, Collector), not the byte[] API.","If you only have raw bytes, construct a ConsumerRecord with a null key/schema and pass it to deserialize."],"exampleFix":"// before\nSeaTunnelRow row = schema.deserialize(message);\n// after\nschema.deserialize(message, new CollectorToList<>(collector));","handlingStrategy":"type-guard","validationCode":"if (schema instanceof CompatibleKafkaConnectDeserializationSchema) { /* must use Collector overload */ }","typeGuard":"boolean needsCollectorOverload = schema instanceof CompatibleKafkaConnectDeserializationSchema;","tryCatchPattern":"try { schema.deserialize(msg, collector); } catch (UnsupportedOperationException e) { /* fall back to record-based path */ }","preventionTips":["Always call the Collector-based overload for Kafka Connect formats","Write tests through the same API path the connector uses"],"tags":["kafka","deserialization","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-14T05:17:10.506Z"}