{"record":{"id":"2a6f6f9460498086","repo":"apache/flink","slug":"corrupt-canal-json-message-s","errorCode":null,"errorMessage":"Corrupt Canal JSON message '%s'.","messagePattern":"Corrupt Canal JSON message '(.+?)'\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java","lineNumber":306,"sourceCode":"                return;\n            } else {\n                if (!ignoreParseErrors) {\n                    throw new IOException(\n                            format(\n                                    \"Unknown \\\"type\\\" value \\\"%s\\\". The Canal JSON message is '%s'\",\n                                    type, new String(message)));\n                }\n                if (LOG.isDebugEnabled()) {\n                    LOG.debug(\n                            \"Unknown \\\"type\\\" value '{}'. The Canal JSON message is '{}'.\",\n                            type,\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 Canal JSON message '%s'.\", new String(message)), t);\n            }\n            if (LOG.isDebugEnabled()) {\n                LOG.debug(\"Corrupt Canal 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        final int physicalArity = physicalRow.getArity();\n        final int metadataArity = metadataConverters.length;","sourceCodeStart":288,"sourceCodeEnd":324,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java#L288-L324","documentation":"Thrown by CanalJsonDeserializationSchema's catch-all when any Throwable escapes the per-message parsing pipeline (JSON syntax error, missing 'data'/'type' fields, unexpected shapes, field conversion failures). It wraps the original exception with the full raw message for diagnosis. With 'json.ignore-parse-errors' enabled it is logged at DEBUG and the message is skipped instead.","triggerScenarios":"Consuming non-JSON bytes, truncated messages, JSON lacking the Canal envelope ('data', 'type', 'old'), or payload values incompatible with the declared table schema, using format 'canal-json' with default options.","commonSituations":"Compacted/corrupted Kafka messages; misconfigured topic with mixed formats (plain JSON, Debezium, or log lines); partial writes from producers; schema drift between Canal-JSON payload and the Flink table columns.","solutions":["Capture and pretty-print the failing raw message from the exception text and validate its Canal envelope (must have data/type, and old for updates)","Fix the producer or topic routing so only well-formed Canal-JSON reaches this table","Align the table schema with the 'data' payload's actual columns","If occasional poison pills are acceptable, set 'json.ignore-parse-errors' = true (records are then skipped, not failed)"],"exampleFix":"// before\nWITH ('connector'='kafka', 'format'='canal-json')\n\n// after: tolerate poison pills (skips bad messages)\nWITH ('connector'='kafka', 'format'='canal-json', 'json.ignore-parse-errors'='true')","handlingStrategy":"try-catch","validationCode":"JsonNode root;\ntry { root = mapper.readTree(message); } catch (IOException e) { /* poison pill: quarantine */ }\nif (root.get(\"type\") == null || root.get(\"data\") == null) { /* not canal-json envelope */ }","typeGuard":null,"tryCatchPattern":"catch (IOException e) on 'Corrupt Canal JSON message' — log raw message + cause to a dead-letter store; enable ignore-parse-errors only with skip-rate monitoring.","preventionTips":["Validate the envelope of a sample message before creating the table","Isolate CDC topics from other producers","Monitor skipped-message counters when ignore-parse-errors is on"],"tags":["canal","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"}