apache/flink · error · RuntimeException

Please invoke DeserializationSchema#deserialize(byte[], Coll

Error message

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

What it means

RuntimeException from DebeziumAvroDeserializationSchema.deserialize(byte[]): this class only supports the collector-based variant, which must output both the 'before' and 'after' rows for UPDATE events and cannot return a single value. The single-argument overload is intentionally blocked.

Source

Thrown at flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java:124

                        producedTypeInfo);
    }

    @VisibleForTesting
    DebeziumAvroDeserializationSchema(
            TypeInformation<RowData> producedTypeInfo,
            AvroRowDataDeserializationSchema avroDeserializer) {
        this.producedTypeInfo = producedTypeInfo;
        this.avroDeserializer = avroDeserializer;
    }

    @Override
    public void open(InitializationContext context) throws Exception {
        avroDeserializer.open(context);
    }

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

    @Override
    public void deserialize(byte[] message, Collector<RowData> out) throws IOException {

        if (message == null || message.length == 0) {
            // skip tombstone messages
            return;
        }
        try {
            GenericRowData row = (GenericRowData) avroDeserializer.deserialize(message);

            GenericRowData before = (GenericRowData) row.getField(0);
            GenericRowData after = (GenericRowData) row.getField(1);
            String op = row.getField(2).toString();
            if (OP_CREATE.equals(op) || OP_READ.equals(op)) {
                after.setRowKind(RowKind.INSERT);

View on GitHub (pinned to 2f3c205e92)

Solutions

Call deserialize(byte[] message, Collector<RowData> out) instead and collect from the collector.
If integrating with an API that requires the one-arg form, use the plain (non-Debezium) AvroRowDataDeserializationSchema instead.

Example fix

// before
RowData row = schema.deserialize(message);

// after
List<RowData> out = new ArrayList<>();
schema.deserialize(message, out::add); // UPDATE emits before+after rows

Defensive patterns

Strategy: type-guard

Type guard

static boolean supportsSingleArgDeserialize(DeserializationSchema<?> d) {
    return !(d instanceof DebeziumAvroDeserializationSchema);
}

Try / catch

try {
    RowData r = schema.deserialize(message);
} catch (RuntimeException e) {
    if (e.getMessage().contains("Collector<RowData>")) {
        // switch to the collector overload: schema.deserialize(message, collector)
    }
    throw e;
}

Prevention

Always call the collector-based deserialize(byte[], Collector) for CDC deserializers.
In tests, collect into a List<RowData> and assert on all emitted rows (UPDATE yields two).

When it happens

Trigger: Calling DeserializationSchema#deserialize(byte[]) on a DebeziumAvroDeserializationSchema instance — e.g. a custom Kafka consumer, a test harness, or a generic framework invoking the one-arg method by default instead of passing a Collector.

Common situations: User code or older connector versions that only call the single-message API; unit tests calling deserialize(message) directly.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/1ec594322c322a0d. Report an issue: GitHub.