apache/flink · error · RuntimeException

Please invoke DeserializationSchema#deserialize(byte[], Coll

Error message

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

What it means

Thrown by DebeziumJsonDeserializationSchema.deserialize(byte[]) because the single-argument DeserializationSchema method is deliberately unsupported: one Debezium message can yield multiple output rows (UPDATE emits before+after) and needs the collector-based variant. Flink's runtime always uses the collector overload; hitting this means the wrong overload was called directly.

Source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java:136

                        ignoreParseErrors,
                        timestampFormat);
        this.hasMetadata = requestedMetadata.size() > 0;
        this.metadataConverters =
                createMetadataConverters(jsonRowType, requestedMetadata, schemaInclude);
        this.producedTypeInfo = producedTypeInfo;
        this.schemaInclude = schemaInclude;
        this.ignoreParseErrors = ignoreParseErrors;
    }

    @Override
    public void open(InitializationContext context) throws Exception {
        genericRowDataList = new ArrayList<>();
        jsonDeserializer.open(context);
    }

    @Override
    public RowData deserialize(byte[] message) {
        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;
        }
        genericRowDataList.clear();
        try {
            GenericRowData row = (GenericRowData) jsonDeserializer.deserialize(message);
            GenericRowData payload;
            if (schemaInclude) {
                payload = (GenericRowData) row.getField(0);
            } else {
                payload = row;
            }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Call deserialize(message, collector) and gather rows from the collector
2. Let Flink's runtime drive the schema (normal Kafka connector usage) instead of manual invocation
3. In tests, use a list-backed collector to capture all emitted rows (UPDATE yields two)

Example fix

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

// after
List<RowData> rows = new ArrayList<>();
schema.deserialize(message, new CollectedListCollector<>(rows));

if (schema instanceof DebeziumJsonDeserializationSchema) {
    schema.deserialize(msg, collector); // one message -> N rows (UPDATE = 2)
}

static boolean needsCollector(DeserializationSchema<RowData> s) {
    return s instanceof DebeziumJsonDeserializationSchema;
}

When it happens

Trigger: Manually invoking new DebeziumJsonDeserializationSchema(...).deserialize(bytes) — in tests, custom sources, or utilities — instead of deserialize(message, Collector<RowData>).

Common situations: Custom source implementations or test harnesses coded against the generic DeserializationSchema interface; code copied from plain-JSON format usage; ad-hoc debugging tools pushing single messages through the schema.

Related errors

• Please invoke DeserializationSchema#deserialize(byte[], Coll
• Please invoke DeserializationSchema#deserialize(byte[], Coll
• The given buffer is neither an array-backed heap ByteBuffer,
• The "before" field of %s message is null, if you are using D
• The "before" field of %s message is null, if you are using D

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