apache/flink · error · IOException

Corrupt Ogg JSON message '%s'.

Error message

Corrupt Ogg JSON message '%s'.

What it means

The protective outer catch in OggJsonDeserializationSchema.deserialize(byte[], Collector) converts any Throwable raised while deserializing the message and dispatching on op_type into IOException('Corrupt Ogg JSON message ...') with the original cause attached (when ignore-parse-errors = false). Note the REPLICA_IDENTITY IllegalStateException from the before-null checks is also caught here and re-wrapped, so check the cause.

Source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java:216

                genericRowDataList.add(emitRow(row, before));
            } else {
                if (!ignoreParseErrors) {
                    throw new IOException(
                            format(
                                    "Unknown \"op_type\" value \"%s\". The Ogg JSON message is '%s'",
                                    op, new String(message)));
                }
                if (LOG.isDebugEnabled()) {
                    LOG.debug(
                            "Unknown \"op_type\" value '{}'. The Ogg 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 Ogg JSON message '%s'.", new String(message)), t);
            }
            if (LOG.isDebugEnabled()) {
                LOG.debug("Corrupt Ogg JSON message '{}'.", new String(message), t);
            }
        }
        for (GenericRowData genericRowData : genericRowDataList) {
            out.collect(genericRowData);
        }
    }

    // --------------------------------------------------------------------------------------------

    private GenericRowData emitRow(GenericRowData rootRow, GenericRowData physicalRow) {
        // shortcut in case no output projection is required
        if (!hasMetadata) {
            return physicalRow;
        }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Unwrap the cause in logs to find the true failure point (parse, structure, or replica-identity).
2. Ensure the consumed topic carries Ogg JSON row-change messages and the DDL matches the payload's 'before'/'after' structure.
3. Enable 'ogg-json.ignore-parse-errors' = true if skipping corrupt records is acceptable for the pipeline.

try { deserializer.deserialize(message, collector); } catch (IOException e) { deadLetter.write(message, e.getCause() != null ? e.getCause() : e); }

When it happens

Trigger: Non-JSON bytes; JSON that does not match the expected [before, after, op_type] structure (missing fields causing NPE/CCE); field values incompatible with the declared table schema; the null-before IllegalStateException wrapped during UPDATE/DELETE handling.

Common situations: Wrong topic content (plain JSON, non-Ogg CDC); DDL mismatches with the Ogg payload; partially delivered or compressed messages consumed as raw bytes.

Related errors

• Corrupt Maxwell JSON message '%s'.
• Please invoke DeserializationSchema#deserialize(byte[], Coll
• JSON format doesn't support failOnMissingField and ignorePar
• Please invoke DeserializationSchema#deserialize(byte[], Coll
• Failed to deserialize JSON '%s'.

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