{"record":{"id":"0cfc935be6b8dfee","repo":"apache/flink","slug":"corrupt-debezium-json-message-s","errorCode":null,"errorMessage":"Corrupt Debezium JSON message '%s'.","messagePattern":"Corrupt Debezium JSON message '(.+?)'\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java","lineNumber":195,"sourceCode":"                genericRowDataList.add(handleRow(row, before));\n            } else {\n                if (!ignoreParseErrors) {\n                    throw new IOException(\n                            format(\n                                    \"Unknown \\\"op\\\" value \\\"%s\\\". The Debezium JSON message is '%s'\",\n                                    op, new String(message)));\n                }\n                if (LOG.isDebugEnabled()) {\n                    LOG.debug(\n                            \"Unknown \\\"op\\\" value '{}'. The Debezium 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 Debezium JSON message '%s'.\", new String(message)), t);\n            }\n            if (LOG.isDebugEnabled()) {\n                LOG.debug(\"Corrupt Debezium JSON message '{}'.\", new String(message), t);\n            }\n        }\n        for (GenericRowData genericRowData : genericRowDataList) {\n            out.collect(genericRowData);\n        }\n    }\n\n    private GenericRowData handleRow(GenericRowData rootRow, GenericRowData physicalRow) {\n        // shortcut in case no output projection is required\n        if (!hasMetadata) {\n            return physicalRow;\n        }\n\n        final int physicalArity = physicalRow.getArity();","sourceCodeStart":177,"sourceCodeEnd":213,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java#L177-L213","documentation":"Thrown by DebeziumJsonDeserializationSchema's catch-all when any Throwable escapes message parsing: invalid JSON, missing payload/before/after/op fields, envelope-shape surprises (schema-included vs not), or field-level conversion errors. It wraps the original exception together with the raw message. With 'json.ignore-parse-errors' enabled it degrades to a DEBUG log and the message is skipped.","triggerScenarios":"Consuming malformed or non-Debezium JSON on a 'debezium-json' topic: broken JSON bytes, missing 'payload', wrong envelope for the configured 'debezium-json.schema-include' value, or payload fields incompatible with the declared table schema.","commonSituations":"Schema-include mismatch (producer emits schema-wrapped messages but table omits 'debezium-json.schema-include'='true' or vice versa); misrouted topics (plain JSON or Canal); corrupted/truncated Kafka messages; schema drift between Debezium payload and the Flink DDL.","solutions":["Match 'debezium-json.schema-include' to the actual message envelope (true if messages carry a 'schema' wrapper)","Inspect the failing raw message from the exception and validate its Debezium envelope and payload types against the DDL","Fix topic routing/producer config so clean Debezium JSON reaches the table","If poison pills must be tolerated, set 'json.ignore-parse-errors' = true (bad messages are skipped silently — monitor skip volume)"],"exampleFix":"// before\nWITH ('connector'='kafka', 'format'='debezium-json')\n\n// after (messages wrapped with schema)\nWITH ('connector'='kafka', 'format'='debezium-json',\n  'debezium-json.schema-include'='true')","handlingStrategy":"try-catch","validationCode":"JsonNode root;\ntry { root = mapper.readTree(message); } catch (IOException e) { /* poison pill: quarantine */ }\nboolean hasSchema = root.has(\"schema\");\n// configure 'debezium-json.schema-include' to match hasSchema before consuming","typeGuard":null,"tryCatchPattern":"catch (IOException e) on 'Corrupt Debezium JSON message' — capture message + cause into a dead-letter store; only then decide skip (ignore-parse-errors) vs fix (schema-include, routing).","preventionTips":["Match 'debezium-json.schema-include' to the producer's envelope","Keep Debezium topics single-purpose","Validate envelope shape with jq on a sample before DDL freeze"],"tags":["debezium","cdc","corrupt-message","kafka","runtime"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}