{"record":{"id":"e016e9f6ffa11709","repo":"apache/flink","slug":"the-before-field-of-s-message-is-null-if-you-a-e016e9","errorCode":null,"errorMessage":"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.","messagePattern":"The \"before\" field of (.+?) message is null, if you are using Ogg Postgres Connector, please check the Postgres table has been set REPLICA IDENTITY to FULL level\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"critical","filePath":"flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java","lineNumber":185,"sourceCode":"    @Override\n    public void deserialize(byte[] message, Collector<RowData> out) throws IOException {\n        if (message == null || message.length == 0) {\n            // skip tombstone messages\n            return;\n        }\n        genericRowDataList.clear();\n        try {\n            GenericRowData row = (GenericRowData) jsonDeserializer.deserialize(message);\n\n            GenericRowData before = (GenericRowData) row.getField(0);\n            GenericRowData after = (GenericRowData) row.getField(1);\n            String op = row.getField(2).toString();\n            if (OP_CREATE.equals(op)) {\n                after.setRowKind(RowKind.INSERT);\n                genericRowDataList.add(emitRow(row, after));\n            } else if (OP_UPDATE.equals(op)) {\n                if (before == null) {\n                    throw new IllegalStateException(\n                            String.format(REPLICA_IDENTITY_EXCEPTION, \"UPDATE\"));\n                }\n                before.setRowKind(RowKind.UPDATE_BEFORE);\n                after.setRowKind(RowKind.UPDATE_AFTER);\n                genericRowDataList.add(emitRow(row, before));\n                genericRowDataList.add(emitRow(row, after));\n            } else if (OP_DELETE.equals(op)) {\n                if (before == null) {\n                    throw new IllegalStateException(\n                            String.format(REPLICA_IDENTITY_EXCEPTION, \"DELETE\"));\n                }\n                before.setRowKind(RowKind.DELETE);\n                genericRowDataList.add(emitRow(row, before));\n            } else {\n                if (!ignoreParseErrors) {\n                    throw new IOException(\n                            format(\n                                    \"Unknown \\\"op_type\\\" value \\\"%s\\\". The Ogg JSON message is '%s'\",","sourceCodeStart":167,"sourceCodeEnd":203,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java#L167-L203","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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)."],"exampleFix":"-- before (source Postgres table, default identity)\nCREATE TABLE orders (...);\n-- after\nALTER TABLE orders REPLICA IDENTITY FULL;","handlingStrategy":"validation","validationCode":"// Pre-check the before image for update events before deserialization\nJsonNode n = MAPPER.readTree(message);\nif (\"U\".equals(n.path(\"op_type\").asText()) && n.path(\"before\").isNull()) {\n    throw new IllegalStateException(\"Source table needs REPLICA IDENTITY FULL; skipping message\");\n}","typeGuard":null,"tryCatchPattern":"try { deserializer.deserialize(m, collector); } catch (IOException e) { if (e.getCause() instanceof IllegalStateException && message mentions REPLICA IDENTITY) alertDba(); throw e; }","preventionTips":["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."],"tags":["flink","ogg","json","cdc","postgres","replica-identity"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}