Hi Team, I am currently facing an issue when using Flink CDC 3.5.0 with Flink 1.20.3 to ingest data from Postgres database table. Attaching sample CDC sql script CREATE TABLE products ( id INT, name STRING, description STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'postgres-cdc', 'hostname' = 'localhost', 'port' = '5433', 'username' = 'pg-connector', 'password' = 'pg123abc', 'database-name' = 'demo', 'schema-name' = 'public', 'table-name' = 'products', 'debezium.publication.name' = 'dbz_publication', 'debezium.plugin.name' = 'pgoutput', 'slot.name' = 'flink_products_slot', 'debezium.heartbeat.action.query' = 'SELECT pg_logical_emit_message(false, ''heartbeat'', now()::varchar)', 'debezium.heartbeat.interval.ms' = '10000' ); Basically when I add the heartbeat check I am getting error as debezium is trying to deserialize "before" field data from topic "*postgres_cdc_source.message*" but the field does not exist. How to handle this from flink side as it is breaking the job? Any suggestions are appreciated 2026-03-05 115455,608 ERROR org.apache.flink.cdc.debezium.internal.DebeziumChangeFetcher [] - Failed to deserialize record SourceRecord{sourcePartition={server=postgres_cdc_source}, sourceOffset={transaction_id=null, lsn_proc=30035760, lsn_commit=30035760, lsn=30035760, ts_usec=1772732890713362}} ConnectRecord{topic='postgres_cdc_source.message', kafkaPartition=null, key=Struct{prefix=heartbeat}, keySchema=Schema{io.debezium.connector.postgresql.MessageKey:STRUCT}, value=Struct{op=m,ts_ms=1772732891698,source=Struct{version=1.9.8.Final,connector=postgresql,name=postgres_cdc_source,ts_ms=1772732890713,db=demo,sequence=["30035760","30035760"],schema=,table=,lsn=30035760},message=Struct{prefix=heartbeat,content=java.nio.HeapByteBuffer[pos=0 lim=28 cap=28]}}, valueSchema=Schema{io.debezium.connector.postgresql.MessageValue:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)} java.lang.NullPointerException: Cannot invoke "org.apache.flink.cdc.connectors.shaded.org.apache.kafka.connect.data.Field.schema()" because the return value of "org.apache.flink.cdc.connectors.shaded.org.apache.kafka.connect.data.Schema.field(String)" is null at org.apache.flink.cdc.debezium.table.RowDataDebeziumDeserializeSchema.extractBeforeRow(RowDataDebeziumDeserializeSchema.java:161) ~[flink-sql-connector-postgres-cdc-3.5-SNAPSHOT.jar:3.5-SNAPSHOT] at org.apache.flink.cdc.debezium.table.RowDataDebeziumDeserializeSchema.deserialize(RowDataDebeziumDeserializeSchema.java:141) ~[flink-sql-connector-postgres-cdc-3.5-SNAPSHOT.jar:3.5-SNAPSHOT] at org.apache.flink.cdc.debezium.internal.DebeziumChangeFetcher.handleBatch(DebeziumChangeFetcher.java:248) [flink-sql-connector-postgres-cdc-3.5-SNAPSHOT.jar:3.5-SNAPSHOT] at org.apache.flink.cdc.debezium.internal.DebeziumChangeFetcher.runFetchLoop(DebeziumChangeFetcher.java:167) [flink-sql-connector-postgres-cdc-3.5-SNAPSHOT.jar:3.5-SNAPSHOT] at org.apache.flink.cdc.debezium.DebeziumSourceFunction.run(DebeziumSourceFunction.java:447) [flink-sql-connector-postgres-cdc-3.5-SNAPSHOT.jar:3.5-SNAPSHOT] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:113) [flink-dist-1.20.3.jar:1.20.3] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:71) [flink-dist-1.20.3.jar:1.20.3] at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:338) [flink-dist-1.20.3.jar:1.20.3] 2026-03-05 115455,625 INFO io.debezium.embedded.EmbeddedEngine [] - Stopping the embedded engine 2026-03-05 115455,627 INFO io.debezium.embedded.EmbeddedEngine [] - Waiting for PT5M for connector to stop