apache/flink · error · IOException

Corrupt Canal JSON message '%s'.

Error message

Corrupt Canal JSON message '%s'.

What it means

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.

Source

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

                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()) {
                LOG.debug("Corrupt Canal JSON message '{}'.", new String(message), t);
            }
        }
        for (GenericRowData genericRowData : genericRowDataList) {
            out.collect(genericRowData);
        }
    }

    private GenericRowData handleRow(GenericRowData rootRow, GenericRowData physicalRow) {
        // shortcut in case no output projection is required
        if (!hasMetadata) {
            return physicalRow;
        }
        final int physicalArity = physicalRow.getArity();
        final int metadataArity = metadataConverters.length;

View on GitHub (pinned to 2f3c205e92)

Solutions

  1. 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)
  2. Fix the producer or topic routing so only well-formed Canal-JSON reaches this table
  3. Align the table schema with the 'data' payload's actual columns
  4. If occasional poison pills are acceptable, set 'json.ignore-parse-errors' = true (records are then skipped, not failed)

Example fix

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

// after: tolerate poison pills (skips bad messages)
WITH ('connector'='kafka', 'format'='canal-json', 'json.ignore-parse-errors'='true')
Defensive patterns

Strategy: try-catch

Validate before calling

JsonNode root;
try { root = mapper.readTree(message); } catch (IOException e) { /* poison pill: quarantine */ }
if (root.get("type") == null || root.get("data") == null) { /* not canal-json envelope */ }

Try / catch

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.

Prevention

When it happens

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

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

Related errors

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