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 CanalJsonDeserializationSchema.deserialize(byte[]) because the single-argument DeserializationSchema method is intentionally unsupported: Canal messages map to multiple RowData records (e.g. UPDATE produces before+after) and need the collector-based variant. The runtime always calls the collector variant; hitting this means someone invoked the wrong overload directly.

Source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java:218

                    producedTypeInfo,
                    database,
                    table,
                    ignoreParseErrors,
                    timestampFormat);
        }
    }

    // ------------------------------------------------------------------------------------------

    @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(@Nullable byte[] message, Collector<RowData> out) throws IOException {
        if (message == null || message.length == 0) {
            return;
        }
        genericRowDataList.clear();
        try {
            final JsonNode root = jsonDeserializer.deserializeToJsonNode(message);
            if (database != null) {
                if (!databasePattern
                        .matcher(root.get(ReadableMetadata.DATABASE.key).asText())
                        .matches()) {
                    return;
                }
            }

View on GitHub (pinned to 2f3c205e92)

Solutions

Switch the call to deserialize(message, collector) and collect results from the collector
In custom sources, implement SourceFunction/reader so Flink's runtime drives deserialization with the collector variant
For tests, collect outputs into a ListCollector and assert on all emitted rows

Example fix

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

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

Defensive patterns

Strategy: type-guard

Validate before calling

if (schema instanceof CanalJsonDeserializationSchema) {
    // must use collector variant: one message -> N rows
    schema.deserialize(msg, collector);
}

Type guard

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

Prevention

In custom sources/tests always code against deserialize(byte[], Collector)
Remember CDC formats are one-to-many: an UPDATE yields two rows

When it happens

Trigger: Manually calling new CanalJsonDeserializationSchema(...).deserialize(bytes) with a single byte[] argument — e.g. in unit tests, custom source functions, or ad-hoc tooling — instead of the two-argument deserialize(byte[], Collector<RowData>).

Common situations: Custom SourceFunction or test harnesses written against the generic DeserializationSchema interface; copy-pasted code from non-CDC formats (plain JSON schema supports the single-arg method); debugging utilities that push single messages through the schema.

Related errors

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