yuxiqian commented on code in PR #4439:
URL: https://github.com/apache/flink-cdc/pull/4439#discussion_r3556932163
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -144,12 +144,42 @@ public TypeInformation<Event> getProducedType() {
private RecordData extractBeforeDataRecord(Struct value, Schema
valueSchema) throws Exception {
Schema beforeSchema = fieldSchema(valueSchema,
Envelope.FieldName.BEFORE);
Struct beforeValue = fieldStruct(value, Envelope.FieldName.BEFORE);
+ if (beforeValue == null) {
+ throw new NullPointerException(
+ "Before data is null for UPDATE/DELETE event. "
+ + formatHint(getNullBeforeDataHint())
+ + "Schema name: "
+ + valueSchema.name());
+ }
Review Comment:
I wonder if we can throw a generic NPE in `extractBeforeDataRecord`, and
wrap it in PostgresEventDeserializer like this?
```java
// 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:
```java
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
