apache/flink · critical · IllegalStateException

The "before" field of %s message is null, if you are using O

Error message

The "before" field of %s message is null, if you are using Ogg Postgres Connector, please check the Postgres table has been set REPLICA IDENTITY to FULL level.

What it means

For an Ogg 'UPDATE' (op_type U) message, the 'before' image deserialized to null. The deserializer must emit an UPDATE_BEFORE row, which requires the before image; the message template (REPLICA_IDENTITY_EXCEPTION) tells Ogg-Postgres users that the source table must use REPLICA IDENTITY FULL for the connector to capture the full before-row.

Source

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

    @Override
    public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
        if (message == null || message.length == 0) {
            // skip tombstone messages
            return;
        }
        genericRowDataList.clear();
        try {
            GenericRowData row = (GenericRowData) jsonDeserializer.deserialize(message);

            GenericRowData before = (GenericRowData) row.getField(0);
            GenericRowData after = (GenericRowData) row.getField(1);
            String op = row.getField(2).toString();
            if (OP_CREATE.equals(op)) {
                after.setRowKind(RowKind.INSERT);
                genericRowDataList.add(emitRow(row, after));
            } else if (OP_UPDATE.equals(op)) {
                if (before == null) {
                    throw new IllegalStateException(
                            String.format(REPLICA_IDENTITY_EXCEPTION, "UPDATE"));
                }
                before.setRowKind(RowKind.UPDATE_BEFORE);
                after.setRowKind(RowKind.UPDATE_AFTER);
                genericRowDataList.add(emitRow(row, before));
                genericRowDataList.add(emitRow(row, after));
            } else if (OP_DELETE.equals(op)) {
                if (before == null) {
                    throw new IllegalStateException(
                            String.format(REPLICA_IDENTITY_EXCEPTION, "DELETE"));
                }
                before.setRowKind(RowKind.DELETE);
                genericRowDataList.add(emitRow(row, before));
            } else {
                if (!ignoreParseErrors) {
                    throw new IOException(
                            format(
                                    "Unknown \"op_type\" value \"%s\". The Ogg JSON message is '%s'",

View on GitHub (pinned to 2f3c205e92)

Solutions

On the Postgres source: ALTER TABLE <table> REPLICA IDENTITY FULL; so updates carry the complete before image.
Check the Ogg connector's before-image capture setting (e.g. include-old-values / getUpdates behavior) and enable old-value delivery.
Skip such messages with 'ogg-json.ignore-parse-errors' = true only if losing the UPDATE_BEFORE side is acceptable for your downstream (breaks changelog correctness for retraction sinks).

Example fix

-- before (source Postgres table, default identity)
CREATE TABLE orders (...);
-- after
ALTER TABLE orders REPLICA IDENTITY FULL;

Defensive patterns

Strategy: validation

Validate before calling

// Pre-check the before image for update events before deserialization
JsonNode n = MAPPER.readTree(message);
if ("U".equals(n.path("op_type").asText()) && n.path("before").isNull()) {
    throw new IllegalStateException("Source table needs REPLICA IDENTITY FULL; skipping message");
}

Try / catch

try { deserializer.deserialize(m, collector); } catch (IOException e) { if (e.getCause() instanceof IllegalStateException && message mentions REPLICA IDENTITY) alertDba(); throw e; }

Prevention

Set ALTER TABLE ... REPLICA IDENTITY FULL on all captured Postgres tables before enabling the Ogg connector.
Monitor for the replica-identity error and alert the DBA rather than silently skipping.

When it happens

Trigger: OggJsonDeserializationSchema.deserialize processes op_type 'U' where row.getField(0) (the 'before' field) is null. Typical when the Postgres table has default REPLICA IDENTITY (only PK columns in the before image) or the connector is configured to omit old values.

Common situations: Ogg Postgres connector capturing updates on a table without REPLICA IDENTITY FULL; mixed workloads where only some tables are configured correctly; first UPDATE after table setup before identity was altered.

Related errors

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