{"record":{"id":"bfc7f28dd594829d","repo":"apache/flink","slug":"corrupt-ogg-json-message-s","errorCode":null,"errorMessage":"Corrupt Ogg JSON message '%s'.","messagePattern":"Corrupt Ogg JSON message '(.+?)'\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java","lineNumber":216,"sourceCode":"                genericRowDataList.add(emitRow(row, before));\n            } else {\n                if (!ignoreParseErrors) {\n                    throw new IOException(\n                            format(\n                                    \"Unknown \\\"op_type\\\" value \\\"%s\\\". The Ogg JSON message is '%s'\",\n                                    op, new String(message)));\n                }\n                if (LOG.isDebugEnabled()) {\n                    LOG.debug(\n                            \"Unknown \\\"op_type\\\" value '{}'. The Ogg JSON message is '{}'.\",\n                            op,\n                            new String(message));\n                }\n            }\n        } catch (Throwable t) {\n            // a big try catch to protect the processing.\n            if (!ignoreParseErrors) {\n                throw new IOException(\n                        format(\"Corrupt Ogg JSON message '%s'.\", new String(message)), t);\n            }\n            if (LOG.isDebugEnabled()) {\n                LOG.debug(\"Corrupt Ogg JSON message '{}'.\", new String(message), t);\n            }\n        }\n        for (GenericRowData genericRowData : genericRowDataList) {\n            out.collect(genericRowData);\n        }\n    }\n\n    // --------------------------------------------------------------------------------------------\n\n    private GenericRowData emitRow(GenericRowData rootRow, GenericRowData physicalRow) {\n        // shortcut in case no output projection is required\n        if (!hasMetadata) {\n            return physicalRow;\n        }","sourceCodeStart":198,"sourceCodeEnd":234,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java#L198-L234","documentation":"The protective outer catch in OggJsonDeserializationSchema.deserialize(byte[], Collector) converts any Throwable raised while deserializing the message and dispatching on op_type into IOException('Corrupt Ogg JSON message ...') with the original cause attached (when ignore-parse-errors = false). Note the REPLICA_IDENTITY IllegalStateException from the before-null checks is also caught here and re-wrapped, so check the cause.","triggerScenarios":"Non-JSON bytes; JSON that does not match the expected [before, after, op_type] structure (missing fields causing NPE/CCE); field values incompatible with the declared table schema; the null-before IllegalStateException wrapped during UPDATE/DELETE handling.","commonSituations":"Wrong topic content (plain JSON, non-Ogg CDC); DDL mismatches with the Ogg payload; partially delivered or compressed messages consumed as raw bytes.","solutions":["Unwrap the cause in logs to find the true failure point (parse, structure, or replica-identity).","Ensure the consumed topic carries Ogg JSON row-change messages and the DDL matches the payload's 'before'/'after' structure.","Enable 'ogg-json.ignore-parse-errors' = true if skipping corrupt records is acceptable for the pipeline."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try { deserializer.deserialize(message, collector); } catch (IOException e) { deadLetter.write(message, e.getCause() != null ? e.getCause() : e); }","preventionTips":["Check the wrapped cause — replica-identity and schema issues hide under this generic message.","Contract-test the topic's message shape before pointing an ogg-json source at it."],"tags":["flink","ogg","json","deserialization","corrupt-data"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}