apache/flink · error · IOException

Corrupt Debezium JSON message '%s'.

Error message

Corrupt Debezium JSON message '%s'.

What it means

Thrown by DebeziumJsonDeserializationSchema's catch-all when any Throwable escapes message parsing: invalid JSON, missing payload/before/after/op fields, envelope-shape surprises (schema-included vs not), or field-level conversion errors. It wraps the original exception together with the raw message. With 'json.ignore-parse-errors' enabled it degrades to a DEBUG log and the message is skipped.

Source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java:195

                genericRowDataList.add(handleRow(row, before));
            } else {
                if (!ignoreParseErrors) {
                    throw new IOException(
                            format(
                                    "Unknown \"op\" value \"%s\". The Debezium JSON message is '%s'",
                                    op, new String(message)));
                }
                if (LOG.isDebugEnabled()) {
                    LOG.debug(
                            "Unknown \"op\" value '{}'. The Debezium JSON message is '{}'.",
                            op,
                            new String(message));
                }
            }
        } catch (Throwable t) {
            // a big try catch to protect the processing.
            if (!ignoreParseErrors) {
                throw new IOException(
                        format("Corrupt Debezium JSON message '%s'.", new String(message)), t);
            }
            if (LOG.isDebugEnabled()) {
                LOG.debug("Corrupt Debezium 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();

View on GitHub (pinned to 2f3c205e92)

Solutions

  1. Match 'debezium-json.schema-include' to the actual message envelope (true if messages carry a 'schema' wrapper)
  2. Inspect the failing raw message from the exception and validate its Debezium envelope and payload types against the DDL
  3. Fix topic routing/producer config so clean Debezium JSON reaches the table
  4. If poison pills must be tolerated, set 'json.ignore-parse-errors' = true (bad messages are skipped silently — monitor skip volume)

Example fix

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

// after (messages wrapped with schema)
WITH ('connector'='kafka', 'format'='debezium-json',
  'debezium-json.schema-include'='true')
Defensive patterns

Strategy: try-catch

Validate before calling

JsonNode root;
try { root = mapper.readTree(message); } catch (IOException e) { /* poison pill: quarantine */ }
boolean hasSchema = root.has("schema");
// configure 'debezium-json.schema-include' to match hasSchema before consuming

Try / catch

catch (IOException e) on 'Corrupt Debezium JSON message' — capture message + cause into a dead-letter store; only then decide skip (ignore-parse-errors) vs fix (schema-include, routing).

Prevention

When it happens

Trigger: Consuming malformed or non-Debezium JSON on a 'debezium-json' topic: broken JSON bytes, missing 'payload', wrong envelope for the configured 'debezium-json.schema-include' value, or payload fields incompatible with the declared table schema.

Common situations: Schema-include mismatch (producer emits schema-wrapped messages but table omits 'debezium-json.schema-include'='true' or vice versa); misrouted topics (plain JSON or Canal); corrupted/truncated Kafka messages; schema drift between Debezium payload and the Flink DDL.

Related errors

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