apache/seatunnel · error · UnsupportedOperationException

Please invoke DeserializationSchema#deserialize(byte[], Coll

Error message

Please invoke DeserializationSchema#deserialize(byte[], Collector<SeaTunnelRow>) instead.

What it means

NativeKafkaConnectDeserializationSchema throws UnsupportedOperationException from deserialize(byte[]) by design. It deserializes Kafka Connect SourceRecords, which require ConsumerRecord context (headers, Connect schema), so only the Collector-based overload is functional.

Source

Thrown at seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/NativeKafkaConnectDeserializationSchema.java:95

    public NativeKafkaConnectDeserializationSchema(
            @NonNull CatalogTable catalogTable,
            boolean keySchemaEnable,
            boolean valueSchemaEnable,
            boolean failOnMissingField,
            boolean ignoreParseErrors) {
        this.catalogTable = catalogTable;
        this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
        this.keySchemaEnable = keySchemaEnable;
        this.valueSchemaEnable = valueSchemaEnable;
        // Runtime converter
        this.runtimeConverter =
                new JsonToRowConverters(failOnMissingField, ignoreParseErrors)
                        .createRowConverter(checkNotNull(seaTunnelRowType));
    }

    @Override
    public SeaTunnelRow deserialize(byte[] message) throws IOException {
        throw new UnsupportedOperationException(
                "Please invoke DeserializationSchema#deserialize(byte[], Collector<SeaTunnelRow>) instead.");
    }

    /**
     * Deserialize kafka consumer record
     *
     * @param msg
     * @param out
     */
    public void deserialize(ConsumerRecord<byte[], byte[]> msg, Collector<SeaTunnelRow> out) {
        tryInitConverter();
        if (msg == null) {
            return;
        }
        Map<String, Object> record = convertToSinkRecord(msg);
        RowKind rowKind = RowKind.INSERT;
        Optional<TablePath> tablePath =
                Optional.ofNullable(catalogTable).map(CatalogTable::getTablePath);

View on GitHub (pinned to cf67b549a7)

Solutions

Use the overload deserialize(ConsumerRecord<byte[], byte[]>, Collector<SeaTunnelRow>).
Run deserialization through SeaTunnel's Kafka source connector, which calls the record-based overload.
Wrap the byte[] payload in a ConsumerRecord (with schema info if schema-registry is enabled) before deserializing.

Example fix

// before
SeaTunnelRow row = schema.deserialize(payload);
// after
schema.deserialize(new ConsumerRecord<>(topic, partition, offset, key, payload), collector);

Defensive patterns

Strategy: type-guard

Validate before calling

if (schema instanceof NativeKafkaConnectDeserializationSchema) { /* use ConsumerRecord overload */ }

Type guard

boolean needsRecordOverload = schema instanceof NativeKafkaConnectDeserializationSchema;

Try / catch

try { schema.deserialize(consumerRecord, collector); } catch (UnsupportedOperationException e) { /* use record-based API */ }

Prevention

Never use deserialize(byte[]) with native Kafka Connect schemas
Route deserialization through the SeaTunnel Kafka connector

When it happens

Trigger: Invoking schema.deserialize(byte[]) directly, or using a caller that exercises the byte[]-only DeserializationSchema entry point instead of deserialize(ConsumerRecord, Collector).

Common situations: Hand-written test harnesses calling the wrong overload; older SeaTunnel connector code paths using the legacy API; user code adapting this schema outside the Kafka connector.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/dba27a7ed703791a. Report an issue: GitHub.