apache/flink · error · IOException

Unknown "type" value "%s". The Canal JSON message is '%s'

Error message

Unknown "type" value "%s". The Canal JSON message is '%s'

What it means

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.

Source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java:291

                    after.setRowKind(RowKind.UPDATE_AFTER);
                    genericRowDataList.add(handleRow(row, before));
                    genericRowDataList.add(handleRow(row, after));
                }
            } else if (OP_DELETE.equals(type)) {
                // "data" field is an array of row, contains deleted rows
                ArrayData data = row.getArray(0);
                for (int i = 0; i < data.size(); i++) {
                    GenericRowData insert = (GenericRowData) data.getRow(i, fieldCount);
                    insert.setRowKind(RowKind.DELETE);
                    genericRowDataList.add(handleRow(row, insert));
                }
            } else if (OP_CREATE.equals(type)) {
                // "data" field is null and "type" is "CREATE" which means
                // this is a DDL change event, and we should skip it.
                return;
            } else {
                if (!ignoreParseErrors) {
                    throw new IOException(
                            format(
                                    "Unknown \"type\" value \"%s\". The Canal JSON message is '%s'",
                                    type, new String(message)));
                }
                if (LOG.isDebugEnabled()) {
                    LOG.debug(
                            "Unknown \"type\" value '{}'. The Canal JSON message is '{}'.",
                            type,
                            new String(message));
                }
            }
        } catch (Throwable t) {
            // a big try catch to protect the processing.
            if (!ignoreParseErrors) {
                throw new IOException(
                        format("Corrupt Canal JSON message '%s'.", new String(message)), t);
            }
            if (LOG.isDebugEnabled()) {

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Set the format option 'json.ignore-parse-errors' = true in the WITH clause to skip unhandled event types
2. Filter the topic upstream so only INSERT/UPDATE/DELETE data events reach Flink (e.g. Canal server filter configuration)
3. If the event type matters, pre-process the stream (Kafka Streams/another consumer) to normalize or drop unknown types
4. Verify the message really is Canal-JSON and not Debezium-JSON on a mislabeled topic

Example fix

// before
WITH ('connector'='kafka', 'topic'='...', 'format'='canal-json')

// after
WITH ('connector'='kafka', 'topic'='...', 'format'='canal-json',
  'json.ignore-parse-errors'='true')

String type = node.get("type").asText();
if (!Set.of("INSERT","UPDATE","DELETE","CREATE").contains(type)) { /* route out or skip */ }

static boolean isHandledCanalType(String t) {
    return t != null && Set.of("INSERT","UPDATE","DELETE","CREATE").contains(t);
}

catch (IOException e) on 'Unknown "type" value' — either enable ignore-parse-errors or fix the topic; retrying the same message rethrows.

When it happens

Trigger: 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'.

Common situations: 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.

Related errors

• Corrupt Canal JSON message '%s'.
• Could not serialize row '%s'.
• Unknown "op" value "%s". The Debezium JSON message is '%s'
• Could not find field with name '%s'.
• Please invoke DeserializationSchema#deserialize(byte[], Coll

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/76c124f05d0b3717. Report an issue: GitHub.