{"record":{"id":"653f8c986048d3e6","repo":"apache/flink","slug":"please-invoke-deserializationschema-deserialize-by-653f8c","errorCode":null,"errorMessage":"Please invoke DeserializationSchema#deserialize(byte[], Collector<RowData>) instead.","messagePattern":"Please invoke DeserializationSchema#deserialize\\(byte\\[\\], Collector<RowData>\\) instead\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonDeserializationSchema.java","lineNumber":130,"sourceCode":"                        timestampFormat);\n        this.hasMetadata = requestedMetadata.size() > 0;\n        this.metadataConverters = createMetadataConverters(jsonRowType, requestedMetadata);\n        this.producedTypeInfo = producedTypeInfo;\n        this.ignoreParseErrors = ignoreParseErrors;\n        final RowType physicalRowType = ((RowType) physicalDataType.getLogicalType());\n        this.fieldNames = physicalRowType.getFieldNames();\n        this.fieldCount = physicalRowType.getFieldCount();\n    }\n\n    @Override\n    public void open(InitializationContext context) throws Exception {\n        genericRowDataList = new ArrayList<>();\n        jsonDeserializer.open(context);\n    }\n\n    @Override\n    public RowData deserialize(byte[] message) throws IOException {\n        throw new RuntimeException(\n                \"Please invoke DeserializationSchema#deserialize(byte[], Collector<RowData>) instead.\");\n    }\n\n    @Override\n    public void deserialize(byte[] message, Collector<RowData> out) throws IOException {\n        if (message == null || message.length == 0) {\n            return;\n        }\n        genericRowDataList.clear();\n        try {\n            final JsonNode root = jsonDeserializer.deserializeToJsonNode(message);\n            final GenericRowData row = (GenericRowData) jsonDeserializer.convertToRowData(root);\n            String type = row.getString(2).toString(); // \"type\" field\n            if (OP_INSERT.equals(type)) {\n                // \"data\" field is a row, contains inserted rows\n                GenericRowData insert = (GenericRowData) row.getRow(0, fieldCount);\n                insert.setRowKind(RowKind.INSERT);\n                genericRowDataList.add(handleRow(row, insert));","sourceCodeStart":112,"sourceCodeEnd":148,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonDeserializationSchema.java#L112-L148","documentation":"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>).","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nRowData row = schema.deserialize(message);\n// after\nList<RowData> rows = new ArrayList<>();\nschema.deserialize(message, new Collector<RowData>() {\n    public void collect(RowData r) { rows.add(r); }\n    public void close() {}\n});","handlingStrategy":"validation","validationCode":"// Route to the collector-based API; never call deserialize(byte[])\nif (schema instanceof MaxwellJsonDeserializationSchema) {\n    // must use deserialize(byte[], Collector<RowData>)\n    schema.deserialize(message, collector);\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["For multi-row CDC formats always program against the Collector overload.","In tests, assert on the collected List<RowData> (updates yield two rows)."],"tags":["flink","maxwell","json","deserialization","api-misuse"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}