apache/flink · error · RuntimeException

Please invoke DeserializationSchema#deserialize(byte[], Coll

Error message

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

What it means

MaxwellJsonDeserializationSchema implements DeserializationSchema but can emit 0, 1, or 2 rows per input message (UPDATE produces an UPDATE_BEFORE and an UPDATE_AFTER row). The single-result variant deserialize(byte[]) therefore cannot represent its output, so it deliberately throws a RuntimeException directing callers to the Collector-based overload deserialize(byte[], Collector<RowData>).

Source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonDeserializationSchema.java:130

                        timestampFormat);
        this.hasMetadata = requestedMetadata.size() > 0;
        this.metadataConverters = createMetadataConverters(jsonRowType, requestedMetadata);
        this.producedTypeInfo = producedTypeInfo;
        this.ignoreParseErrors = ignoreParseErrors;
        final RowType physicalRowType = ((RowType) physicalDataType.getLogicalType());
        this.fieldNames = physicalRowType.getFieldNames();
        this.fieldCount = physicalRowType.getFieldCount();
    }

    @Override
    public void open(InitializationContext context) throws Exception {
        genericRowDataList = new ArrayList<>();
        jsonDeserializer.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) {
            return;
        }
        genericRowDataList.clear();
        try {
            final JsonNode root = jsonDeserializer.deserializeToJsonNode(message);
            final GenericRowData row = (GenericRowData) jsonDeserializer.convertToRowData(root);
            String type = row.getString(2).toString(); // "type" field
            if (OP_INSERT.equals(type)) {
                // "data" field is a row, contains inserted rows
                GenericRowData insert = (GenericRowData) row.getRow(0, fieldCount);
                insert.setRowKind(RowKind.INSERT);
                genericRowDataList.add(handleRow(row, insert));

View on GitHub (pinned to 2f3c205e92)

Solutions

Switch to deserialize(byte[], Collector<RowData>) and collect from the collector.
In tests, use a CollectingCollector (e.g. a simple Collector<RowData> collecting into a List) to assert on all emitted rows.

Example fix

// before
RowData row = schema.deserialize(message);
// after
List<RowData> rows = new ArrayList<>();
schema.deserialize(message, new Collector<RowData>() {
    public void collect(RowData r) { rows.add(r); }
    public void close() {}
});

Defensive patterns

Strategy: validation

Validate before calling

// Route to the collector-based API; never call deserialize(byte[])
if (schema instanceof MaxwellJsonDeserializationSchema) {
    // must use deserialize(byte[], Collector<RowData>)
    schema.deserialize(message, collector);
}

Prevention

For multi-row CDC formats always program against the Collector overload.
In tests, assert on the collected List<RowData> (updates yield two rows).

When it happens

Trigger: Calling Debezium/Maxwell deserializationSchema.deserialize(message) directly — e.g. in a custom source, a unit test, or old code written against the single-row API. Flink's runtime itself always uses the Collector variant.

Common situations: Unit tests invoking deserialize(byte[]) out of habit; custom SourceReaders reusing the format outside the table connector; code migrated from the plain JSON format whose single-row deserialize works fine.

Related errors

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