{"record":{"id":"898e656c582adda7","repo":"apache/flink","slug":"please-invoke-deserializationschema-deserialize-by-898e65","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/ogg/OggJsonDeserializationSchema.java","lineNumber":163,"sourceCode":"            public Object convert(GenericRowData root, int unused) {\n                return metadata.converter.convert(root, pos);\n            }\n        };\n    }\n\n    private static int findFieldPos(ReadableMetadata metadata, RowType jsonRowType) {\n        return jsonRowType.getFieldNames().indexOf(metadata.requiredJsonField.getName());\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) {\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            // skip tombstone messages\n            return;\n        }\n        genericRowDataList.clear();\n        try {\n            GenericRowData row = (GenericRowData) jsonDeserializer.deserialize(message);\n\n            GenericRowData before = (GenericRowData) row.getField(0);\n            GenericRowData after = (GenericRowData) row.getField(1);\n            String op = row.getField(2).toString();\n            if (OP_CREATE.equals(op)) {\n                after.setRowKind(RowKind.INSERT);","sourceCodeStart":145,"sourceCodeEnd":181,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java#L145-L181","documentation":"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.","triggerScenarios":"Directly calling oggDeserializationSchema.deserialize(message) in custom sources or tests instead of the Collector overload; Flink's runtime never hits this path.","commonSituations":"Custom GoldenGate (Ogg) Kafka consumers reusing the format class; unit tests written against the single-row API; code ported from the plain JSON format.","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)."],"exampleFix":"// before\nRowData row = deserializer.deserialize(bytes);\n// after\nList<RowData> out0 = new ArrayList<>();\ndeserializer.deserialize(bytes, r -> out0.add(r)); // or an anonymous Collector","handlingStrategy":"validation","validationCode":"// Always use the collector overload for Ogg\nList<RowData> emitted = new ArrayList<>();\ndeserializer.deserialize(message, new Collector<RowData>() {\n    @Override public void collect(RowData r) { emitted.add(r); }\n    @Override public void close() {}\n});","typeGuard":null,"tryCatchPattern":null,"preventionTips":["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."],"tags":["flink","ogg","json","deserialization","api-misuse"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}