{"record":{"id":"5cb7ac7d315abcf2","repo":"apache/flink","slug":"the-before-field-of-s-message-is-null-if-you-a-5cb7ac","errorCode":null,"errorMessage":"The \"before\" field of %s message is null, if you are using Debezium 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 Debezium 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/debezium/DebeziumJsonDeserializationSchema.java","lineNumber":164,"sourceCode":"        genericRowDataList.clear();\n        try {\n            GenericRowData row = (GenericRowData) jsonDeserializer.deserialize(message);\n            GenericRowData payload;\n            if (schemaInclude) {\n                payload = (GenericRowData) row.getField(0);\n            } else {\n                payload = row;\n            }\n\n            GenericRowData before = (GenericRowData) payload.getField(0);\n            GenericRowData after = (GenericRowData) payload.getField(1);\n            String op = payload.getField(2).toString();\n            if (OP_CREATE.equals(op) || OP_READ.equals(op)) {\n                after.setRowKind(RowKind.INSERT);\n                genericRowDataList.add(handleRow(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(handleRow(row, before));\n                genericRowDataList.add(handleRow(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(handleRow(row, before));\n            } else {\n                if (!ignoreParseErrors) {\n                    throw new IOException(\n                            format(\n                                    \"Unknown \\\"op\\\" value \\\"%s\\\". The Debezium JSON message is '%s'\",","sourceCodeStart":146,"sourceCodeEnd":182,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java#L146-L182","documentation":"Thrown by DebeziumJsonDeserializationSchema when an UPDATE operation arrives with a null \"before\" field. The format treats this as a misconfigured source: Debezium's Postgres connector only emits full before-images when the table uses REPLICA IDENTITY FULL; otherwise UPDATE before can be null, and Flink cannot build the UPDATE_BEFORE row. The template names the operation ('UPDATE') and points at the replica identity fix.","triggerScenarios":"Debezium Postgres connector capturing a table whose REPLICA IDENTITY is DEFAULT (or NOTHING), producing an UPDATE event with before = null, consumed via format 'debezium-json'. Fails the job at runtime on the first such record.","commonSituations":"Newly captured Postgres tables not altered to REPLICA IDENTITY FULL; defaults on managed Postgres (RDS/CloudSQL) where identity is DEFAULT; switching a Debezium topic from append-only tooling to Flink changelog consumption.","solutions":["On the Postgres source table run: ALTER TABLE <table> REPLICA IDENTITY FULL; then let Debezium capture subsequent updates","If the sink only needs the after-image, use an append-only style consumption or filter so UPDATE handling does not require before","As a stopgap, 'debezium-json' with ignore-parse-errors skips the failing message — but data will be silently lost, so prefer the replica identity fix","Verify with a sample message (jq '.payload.before') that before is populated after the change"],"exampleFix":"-- before\nCREATE TABLE orders (...);  -- REPLICA IDENTITY DEFAULT\n\n-- after\nALTER TABLE orders REPLICA IDENTITY FULL;","handlingStrategy":"validation","validationCode":"// In Postgres, verify before relying on before-images for updates:\n// SELECT relreplident FROM pg_class WHERE relname = '<table>';  -- want 'f' (FULL)\n// If not 'f': ALTER TABLE <table> REPLICA IDENTITY FULL;","typeGuard":null,"tryCatchPattern":"catch (IllegalStateException e) on REPLICA_IDENTITY_EXCEPTION — stop, fix the source table's replica identity, and restart from the checkpoint; the record is not retryable as-is.","preventionTips":["Set REPLICA IDENTITY FULL on every CDC-captured Postgres table at provisioning time","Check relreplident in infrastructure checks","Never paper over this with ignore-parse-errors (silent data loss)"],"tags":["debezium","postgres","replica-identity","cdc","update"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}