[FLINK-38647][CDC] Improve error message for null before/after struct…

[BreadS00]

When PostgreSQL table uses default REPLICA IDENTITY DEFAULT or REPLICA IDENTITY PRIMARY KEY, the logical decoding plugin (pgoutput) may not provide complete before-data for UPDATE events. This causes the before struct to be null in Debezium events, leading to an unhelpful NullPointerException.

[Copilot]

Pull request overview

Improves diagnostics when Debezium before/after payload structs are unexpectedly null (common with PostgreSQL REPLICA IDENTITY DEFAULT/PRIMARY KEY), replacing an unhelpful downstream NullPointerException with a targeted message and an optional connector-specific hint.

Changes:

• Add explicit null checks for before/after structs in DebeziumEventDeserializationSchema, throwing with a clearer message.
• Introduce an overridable getNullBeforeDataHint() hook for connector-specific guidance.
• Provide a PostgreSQL-specific hint suggesting REPLICA IDENTITY FULL.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java	Adds null checks for before/after and a hook for connector-specific hint text.
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializer.java	Overrides the hint hook to provide a PostgreSQL-specific remediation suggestion.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[yuxiqian]

Thanks for @BreadS00's nice work, just left some minor comments.

[BreadS00]

Thanks @yuxiqian for the review! Could you take another look when you have time?

[yuxiqian]

Thank you for your rapid response! Just left some minor comments.

[yuxiqian]

🤕 CI is failing...

Error:  Tests run: 10, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 78.403 s <<< FAILURE! - in org.apache.flink.cdc.connectors.postgres.source.PostgresPipelineITCase
Error:  org.apache.flink.cdc.connectors.postgres.source.PostgresPipelineITCase.testUpdateWithoutReplicaIdentityFullThrowsHelpfulNPE  Time elapsed: 1.15 s  <<< ERROR!
java.lang.IllegalStateException: No operators defined in streaming topology. Cannot execute.
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraphGenerator(StreamExecutionEnvironment.java:2556)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2523)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2514)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2500)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2449)
	at org.apache.flink.cdc.connectors.postgres.source.PostgresPipelineITCase.testUpdateWithoutReplicaIdentityFullThrowsHelpfulNPE(PostgresPipelineITCase.java:1255)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)

[BreadS00]

@yuxiqian ,thank you very much. I've adjusted the test case, it should be fine now.

[yuxiqian]

+1 from my side.

Would @leonardBang and @loserwang1024 like to take another look?

[leonardBang]

Thanks @BreadS00 for the contribution and @yuxiqian for the review, LGTM