{"record":{"id":"76c124f05d0b3717","repo":"apache/flink","slug":"unknown-type-value-s-the-canal-json-message","errorCode":null,"errorMessage":"Unknown \"type\" value \"%s\". The Canal JSON message is '%s'","messagePattern":"Unknown \"type\" value \"(.+?)\"\\. The Canal JSON message is '(.+?)'","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":291,"sourceCode":"                    after.setRowKind(RowKind.UPDATE_AFTER);\n                    genericRowDataList.add(handleRow(row, before));\n                    genericRowDataList.add(handleRow(row, after));\n                }\n            } else if (OP_DELETE.equals(type)) {\n                // \"data\" field is an array of row, contains deleted rows\n                ArrayData data = row.getArray(0);\n                for (int i = 0; i < data.size(); i++) {\n                    GenericRowData insert = (GenericRowData) data.getRow(i, fieldCount);\n                    insert.setRowKind(RowKind.DELETE);\n                    genericRowDataList.add(handleRow(row, insert));\n                }\n            } else if (OP_CREATE.equals(type)) {\n                // \"data\" field is null and \"type\" is \"CREATE\" which means\n                // this is a DDL change event, and we should skip it.\n                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()) {","sourceCodeStart":273,"sourceCodeEnd":309,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java#L273-L309","documentation":"Thrown by CanalJsonDeserializationSchema when a Canal JSON message carries a \"type\" field that is not one of the handled values (INSERT/UPDATE/DELETE/CREATE-with-null-data). With 'json.ignore-parse-errors' disabled (default) this IOException fails the job; with it enabled the message is logged at DEBUG and skipped.","triggerScenarios":"A Canal-JSON topic containing events with type values the deserializer does not handle, such as 'QUERY', 'TRANSACTION', 'HEARTBEAT', 'REPLACE', or unknown vendor extensions, consumed with default options via format 'canal-json'.","commonSituations":"Canal server configured to emit DDL/QUERY/transaction events into the same topic; version differences in Canal's event vocabulary; mixed topics; testing against synthetic Canal JSON with wrong type strings.","solutions":["Set the format option 'json.ignore-parse-errors' = true in the WITH clause to skip unhandled event types","Filter the topic upstream so only INSERT/UPDATE/DELETE data events reach Flink (e.g. Canal server filter configuration)","If the event type matters, pre-process the stream (Kafka Streams/another consumer) to normalize or drop unknown types","Verify the message really is Canal-JSON and not Debezium-JSON on a mislabeled topic"],"exampleFix":"// before\nWITH ('connector'='kafka', 'topic'='...', 'format'='canal-json')\n\n// after\nWITH ('connector'='kafka', 'topic'='...', 'format'='canal-json',\n  'json.ignore-parse-errors'='true')","handlingStrategy":"fallback","validationCode":"String type = node.get(\"type\").asText();\nif (!Set.of(\"INSERT\",\"UPDATE\",\"DELETE\",\"CREATE\").contains(type)) { /* route out or skip */ }","typeGuard":"static boolean isHandledCanalType(String t) {\n    return t != null && Set.of(\"INSERT\",\"UPDATE\",\"DELETE\",\"CREATE\").contains(t);\n}","tryCatchPattern":"catch (IOException e) on 'Unknown \"type\" value' — either enable ignore-parse-errors or fix the topic; retrying the same message rethrows.","preventionTips":["Configure Canal server filters to data events only","Set ignore-parse-errors deliberately with skip monitoring","Contract-test the topic's event-type vocabulary before onboarding"],"tags":["canal","cdc","unknown-type","option","runtime"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}