apache/flink · error · FlinkRuntimeException

Please invoke DeserializationSchema#deserialize(byte[], Coll

Error message

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

What it means

FlinkRuntimeException thrown by the deprecated single-row deserialize(byte[]) path of AbstractJsonDeserializationSchema when a single input message produced more than one RowData. The parser-based JSON deserializer can emit multiple rows per message (e.g., a JSON array '[{...},{...}]' is exploded into one row per element), which cannot be returned from the single-value method, so it fails and tells you to use the Collector variant.

Source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java:125

        if (hasDecimalType) {
            objectMapper.enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
        }
        reusableCollectList = new ArrayList<>();
        collector = new ListCollector<>(reusableCollectList);
    }

    /**
     * @deprecated Use {@link DeserializationSchema#deserialize(byte[], Collector)} instead. The
     *     implementation of {@link AbstractJsonDeserializationSchema#deserialize(byte[])} will be
     *     removed in <a href="https://issues.apache.org/jira/browse/FLINK-37707">FLINK-37707</a>.
     */
    @Deprecated
    @Override
    public RowData deserialize(@Nullable byte[] message) throws IOException {
        reusableCollectList.clear();
        deserialize(message, collector);
        if (reusableCollectList.size() > 1) {
            throw new FlinkRuntimeException(
                    "Please invoke "
                            + "DeserializationSchema#deserialize(byte[], Collector<RowData>) instead.");
        }
        if (reusableCollectList.isEmpty()) {
            return null;
        }
        return reusableCollectList.get(0);
    }

    @Override
    public boolean isEndOfStream(RowData nextElement) {
        return false;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return resultTypeInfo;
    }

View on GitHub (pinned to 2f3c205e92)

Solutions

Switch to deserialize(byte[] message, Collector<RowData> out) which handles multi-row messages
Or wrap the schema in a DeserializationSchema-based caller that always uses the Collector form (as the Kafka/FileSystem connectors do)
If you truly need one-row-per-call semantics, feed single JSON objects rather than arrays

Example fix

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

// after
schema.deserialize(messageBytes, new Collector<RowData>() {
    @Override public void collect(RowData r) { downstream.accept(r); }
    @Override public void close() {}
});

Defensive patterns

Strategy: try-catch

Try / catch

try {
    RowData row = schema.deserialize(bytes);
} catch (FlinkRuntimeException e) {
    if (e.getMessage().contains("Collector")) {
        // switch caller to deserialize(bytes, collector)
    }
}

Prevention

Always call the Collector overload in custom sources/functions
Never assume one byte[] equals one row for JSON: arrays may explode into many

When it happens

Trigger: Calling DeserializationSchema.deserialize(byte[]) on JsonParserRowDataDeserializationSchema when the message is a JSON array with 2+ elements, or any input where the collector receives multiple collect() calls; typical when reusing the schema inside a custom SourceFunction or MapFunction instead of a standard connector runtime.

Common situations: Legacy custom sources that call the deprecated method; Kafka connector is fine (it uses the Collector overload); consuming newline-concatenated or array-form JSON payloads with the parser schema.

Related errors

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