{"record":{"id":"84398a80c03818a3","repo":"apache/seatunnel","slug":"null-data-value-jsonnode-cannot-send-downstre","errorCode":null,"errorMessage":"Null data value '${jsonNode}' Cannot send downstream","messagePattern":"Null data value '(.+?)' Cannot send downstream","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/canal/CanalJsonDeserializationSchema.java","lineNumber":150,"sourceCode":"            if (database != null\n                    && !databasePattern.matcher(jsonNode.get(FIELD_DATABASE).asText()).matches()) {\n                return;\n            }\n            if (table != null\n                    && !tablePattern.matcher(jsonNode.get(FIELD_TABLE).asText()).matches()) {\n                return;\n            }\n\n            JsonNode dataNode = jsonNode.get(FIELD_DATA);\n            String op = jsonNode.get(FIELD_TYPE).asText();\n            JsonNode tsNode = jsonNode.get(FIELD_TS);\n            // When a null value is encountered, an exception needs to be thrown for easy sensing\n            if (dataNode == null || dataNode.isNull()) {\n                // We'll skip the query or create or alter event data\n                if (OP_QUERY.equals(op) || OP_CREATE.equals(op) || OP_ALTER.equals(op)) {\n                    return;\n                }\n                throw new IllegalStateException(\n                        format(\"Null data value '%s' Cannot send downstream\", jsonNode));\n            }\n\n            switch (op) {\n                case OP_INSERT:\n                    for (int i = 0; i < dataNode.size(); i++) {\n                        SeaTunnelRow row = convertJsonNode(dataNode.get(i));\n                        if (tablePath != null && !tablePath.toString().isEmpty()) {\n                            row.setTableId(tablePath.toString());\n                        }\n                        if (tsNode != null) {\n                            MetadataUtil.setEventTime(row, tsNode.asLong());\n                        }\n                        out.collect(row);\n                    }\n                    break;\n                case OP_UPDATE:\n                    final ArrayNode oldNode = (ArrayNode) jsonNode.get(FIELD_OLD);","sourceCodeStart":132,"sourceCodeEnd":168,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/canal/CanalJsonDeserializationSchema.java#L132-L168","documentation":"When a Canal JSON event carries a null or JSON-null 'data' node, the deserializer normally skips query/create/alter DDL events, but for any other op type a null data value cannot produce a row, so it throws IllegalStateException to make the malformed/unsupported event visible rather than emitting an empty row downstream.","triggerScenarios":"Consuming a Canal (MySQL binlog) JSON message whose op is not QUERY/CREATE/ALTER but whose data field is null or null-literal (e.g. heartbeats, some DELETE/other events with no data payload).","commonSituations":"Canal server heartbeat or control messages mixed into the monitored queue/topic; binlog events such as ROTATE or unsupported event types forwarded by Canal; filtering not applied upstream on message types.","solutions":["Filter out non-data events (heartbeat/ROTATE/control events) before deserialization, e.g. only pass op INSERT/UPDATE/DELETE messages","Skip events where data is null in custom logic by pre-checking the JSON, or patch the deserializer to ignore them like QUERY/CREATE/ALTER","Check Canal server configuration to stop emitting heartbeat/control messages to the consumed channel","Set table/database pattern filters (tablePattern/databasePattern) so unrelated events don't reach this code path"],"exampleFix":"// before\n{\"data\":null,\"op\":\"UNKNOWN\",\"es\":1699999999,...}\n// after (pre-filter)\nif (node.path(\"op\").asText(\"\").matches(\"INSERT|UPDATE|DELETE\")\n    && !node.path(\"data\").isNull()) {\n  canalJsonSchema.deserialize(bytes, out);\n}","handlingStrategy":"validation","validationCode":"JsonNode op = node.path(\"op\");\nJsonNode data = node.path(\"data\");\nboolean isDdl = op.asText(\"\").matches(\"QUERY|CREATE|ALTER\");\nif (!isDdl && (data.isMissingNode() || data.isNull())) {\n    return; // skip non-data event before deserializing\n}","typeGuard":"boolean hasData = node.has(\"data\") && !node.get(\"data\").isNull() && node.get(\"data\").isArray();","tryCatchPattern":"try { canalJsonSchema.deserialize(message, out); } catch (IllegalStateException e) { if (e.getMessage().startsWith(\"Null data value\")) { /* skip heartbeat/control event */ } else { throw e; } }","preventionTips":["Filter heartbeat/DDL/control events before deserialization","Configure Canal to not forward control messages to the consumed queue","Apply database/table pattern filters so only data events are consumed","Monitor for Canal version changes that alter event payloads"],"tags":["json","canal","cdc","null-value","malformed-event"],"backgroundTag":"unexpected-api-response-shape","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}