apache/flink · error · RuntimeException

Please invoke DeserializationSchema#deserialize(byte[], Coll

Error message

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

What it means

OggJsonDeserializationSchema may emit one or two rows per message (UPDATE emits an UPDATE_BEFORE and an UPDATE_AFTER row), so the single-row DeserializationSchema#deserialize(byte[]) overload is unimplemented and always throws a RuntimeException pointing callers to deserialize(byte[], Collector<RowData>). This is by design, not a runtime data failure.

Source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java:163

            public Object convert(GenericRowData root, int unused) {
                return metadata.converter.convert(root, pos);
            }
        };
    }

    private static int findFieldPos(ReadableMetadata metadata, RowType jsonRowType) {
        return jsonRowType.getFieldNames().indexOf(metadata.requiredJsonField.getName());
    }

    @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 before = (GenericRowData) row.getField(0);
            GenericRowData after = (GenericRowData) row.getField(1);
            String op = row.getField(2).toString();
            if (OP_CREATE.equals(op)) {
                after.setRowKind(RowKind.INSERT);

View on GitHub (pinned to 2f3c205e92)

Solutions

Use deserialize(byte[], Collector<RowData>) and drain the collector.
In tests, pass a collecting Collector and assert on the full emitted list (2 rows for updates).

Example fix

// before
RowData row = deserializer.deserialize(bytes);
// after
List<RowData> out0 = new ArrayList<>();
deserializer.deserialize(bytes, r -> out0.add(r)); // or an anonymous Collector

Defensive patterns

Strategy: validation

Validate before calling

// Always use the collector overload for Ogg
List<RowData> emitted = new ArrayList<>();
deserializer.deserialize(message, new Collector<RowData>() {
    @Override public void collect(RowData r) { emitted.add(r); }
    @Override public void close() {}
});

Prevention

Treat single-row deserialize(byte[]) as unavailable for all multi-row CDC formats (Maxwell, Ogg, Canal, Debezium).
Wrap format usage behind a helper that always uses the Collector API.

When it happens

Trigger: Directly calling oggDeserializationSchema.deserialize(message) in custom sources or tests instead of the Collector overload; Flink's runtime never hits this path.

Common situations: Custom GoldenGate (Ogg) Kafka consumers reusing the format class; unit tests written against the single-row API; code ported from the plain JSON format.

Related errors

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