{"record":{"id":"8ff8a2b88f711662","repo":"apache/flink","slug":"please-invoke-deserializationschema-deserialize-by-8ff8a2","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/canal/CanalJsonDeserializationSchema.java","lineNumber":218,"sourceCode":"                    producedTypeInfo,\n                    database,\n                    table,\n                    ignoreParseErrors,\n                    timestampFormat);\n        }\n    }\n\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(@Nullable 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            if (database != null) {\n                if (!databasePattern\n                        .matcher(root.get(ReadableMetadata.DATABASE.key).asText())\n                        .matches()) {\n                    return;\n                }\n            }","sourceCodeStart":200,"sourceCodeEnd":236,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java#L200-L236","documentation":"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.","triggerScenarios":"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>).","commonSituations":"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.","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"],"exampleFix":"// before\nRowData row = schema.deserialize(message);\n\n// after\nList<RowData> rows = new ArrayList<>();\nschema.deserialize(message, new CollectedListCollector<>(rows));","handlingStrategy":"type-guard","validationCode":"if (schema instanceof CanalJsonDeserializationSchema) {\n    // must use collector variant: one message -> N rows\n    schema.deserialize(msg, collector);\n}","typeGuard":"static boolean needsCollector(DeserializationSchema<RowData> s) {\n    return s instanceof CanalJsonDeserializationSchema\n        || s instanceof DebeziumJsonDeserializationSchema;\n}","tryCatchPattern":null,"preventionTips":["In custom sources/tests always code against deserialize(byte[], Collector)","Remember CDC formats are one-to-many: an UPDATE yields two rows"],"tags":["canal","cdc","api-misuse","deserialization"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}