[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.

[yuxiqian]

I wonder if we can throw a generic NPE in extractBeforeDataRecord, and wrap it in PostgresEventDeserializer like this?

// In PG event deserializer

@Override
private RecordData extractBeforeDataRecord(Struct value, Schema valueSchema) throws Exception {
    try {
        return super.extractBeforeDataRecord(value, valueSchema);
    } catch (NullPointerException ape) {
        // throw a wrapper exception here, indicating that one may forget to specify REPLICA IDENTITY PRIMARY mode
    }
}

or add extra checking like this:

private RecordData extractBeforeDataRecord(Struct value, Schema valueSchema) throws Exception {
    Struct beforeValue = fieldStruct(value, Envelope.FieldName.BEFORE);
    // Ensure we got proper UPDATE_BEFORE here
    Preconditions.checkNotNull(beforeValue);

    return super.extractBeforeDataRecord(value, valueSchema);
}

WDYT?

[yuxiqian]

I wonder if it's necessary to introduce new methods here just for error messages.