{"record":{"id":"736d5627e57a2884","repo":"apache/flink","slug":"the-before-field-of-s-message-is-null-if-you-a","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":"error","filePath":"flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java","lineNumber":146,"sourceCode":"    @Override\n    public void deserialize(byte[] message, Collector<RowData> out) throws IOException {\n\n        if (message == null || message.length == 0) {\n            // skip tombstone messages\n            return;\n        }\n        try {\n            GenericRowData row = (GenericRowData) avroDeserializer.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) || OP_READ.equals(op)) {\n                after.setRowKind(RowKind.INSERT);\n                out.collect(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                out.collect(before);\n                out.collect(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                out.collect(before);\n            } else {\n                throw new IOException(\n                        format(\n                                \"Unknown \\\"op\\\" value \\\"%s\\\". The Debezium Avro message is '%s'\",\n                                op, new String(message)));","sourceCodeStart":128,"sourceCodeEnd":164,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java#L128-L164","documentation":"IllegalStateException from DebeziumAvroDeserializationSchema: an UPDATE (or DELETE) message arrived whose 'before' field is null. Debezium's Postgres connector only populates 'before' when the table's REPLICA IDENTITY is FULL; otherwise UPDATE/DELETE rows carry no before-image and Flink cannot emit the required UPDATE_BEFORE/DELETE row, so processing fails.","triggerScenarios":"Consuming Debezium Postgres CDC (avro-confluent-registry + debezium-avro) where ALTER TABLE ... REPLICA IDENTITY has not been set to FULL; tables created after the connector started with default replica identity.","commonSituations":"Newly CDC-enabled Postgres tables default to DEFAULT replica identity (only primary key in before-image, null for non-PK or when key missing); DBAs skipping the replica identity step in deployment runbooks.","solutions":["Run: ALTER TABLE <table> REPLICA IDENTITY FULL; on the source Postgres table, then reprocess affected messages.","New tables need the same ALTER after creation — automate it in your migration tooling.","Ensure the Debezium connector has REPLICA IDENTITY rights (owner or pg_role with replica)."],"exampleFix":"-- before (default)\n-- table ships only PK in before-image; UPDATE with null before fails\n\n-- after\nALTER TABLE orders REPLICA IDENTITY FULL;","handlingStrategy":"validation","validationCode":"// before starting the Flink job, verify source table replica identity\n// SELECT relreplident FROM pg_class WHERE relname = 'orders';\n// 'd' = DEFAULT (unsafe), 'f' = FULL (required)\nif (!\"f\".equals(queryPg(\"SELECT relreplident FROM pg_class WHERE relname='orders'\"))) {\n    throw new IllegalStateException(\n        \"orders must be ALTER TABLE orders REPLICA IDENTITY FULL before Debezium CDC\");\n}","typeGuard":"boolean replicaIdentityFull(String replIdent) { return \"f\".equals(replIdent); }","tryCatchPattern":"try {\n    consumer.run();\n} catch (IllegalStateException e) {\n    if (e.getMessage().contains(\"REPLICA IDENTITY\")) {\n        // source-side fix required: ALTER TABLE <t> REPLICA IDENTITY FULL; then reprocess\n        log.error(\"Postgres replica identity too weak for CDC; before-image missing\", e);\n    }\n    throw e;\n}","preventionTips":["Set REPLICA IDENTITY FULL on every CDC-enabled Postgres table at provisioning time (Terraform/Flyway step).","Add a pre-deploy SQL check on pg_class.relreplident for all captured tables.","Monitor for schema-drift: new tables added to the Debezium capture list need the same ALTER."],"tags":["avro","debezium","postgres","cdc","replica-identity","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}