BreadS00 commented on code in PR #4439:
URL: https://github.com/apache/flink-cdc/pull/4439#discussion_r3575714477
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -141,15 +142,23 @@ public TypeInformation<Event> getProducedType() {
return new EventTypeInfo();
}
- private RecordData extractBeforeDataRecord(Struct value, Schema
valueSchema) throws Exception {
+ protected RecordData extractBeforeDataRecord(Struct value, Schema
valueSchema)
+ throws Exception {
Schema beforeSchema = fieldSchema(valueSchema,
Envelope.FieldName.BEFORE);
Struct beforeValue = fieldStruct(value, Envelope.FieldName.BEFORE);
+ Preconditions.checkNotNull(
+ beforeValue,
+ "Before data is null for UPDATE/DELETE event. Schema name: " +
valueSchema.name());
return extractDataRecord(beforeValue, beforeSchema);
}
private RecordData extractAfterDataRecord(Struct value, Schema
valueSchema) throws Exception {
Schema afterSchema = fieldSchema(valueSchema,
Envelope.FieldName.AFTER);
Struct afterValue = fieldStruct(value, Envelope.FieldName.AFTER);
+ Preconditions.checkNotNull(
+ afterValue,
+ "After data is null for CREATE/READ/UPDATE event. Schema name:
"
+ + valueSchema.name());
Review Comment:
Thanks! The only remaining change in the base class is making
extractBeforeDataRecord protected so PostgresEventDeserializer can override it.
The method body is untouched — all the REPLICA IDENTITY logic stays in
PostgresEventDeserializer only.
--
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]
