yuxiqian commented on code in PR #4439:
URL: https://github.com/apache/flink-cdc/pull/4439#discussion_r3569490358
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializer.java:
##########
@@ -83,6 +85,19 @@ public PostgresEventDeserializer(
this.databaseName = databaseName;
}
+ @Override
+ protected RecordData extractBeforeDataRecord(Struct value, Schema
valueSchema)
+ throws Exception {
+ Struct beforeValue = fieldStruct(value, Envelope.FieldName.BEFORE);
+ Preconditions.checkNotNull(
+ beforeValue,
+ "Before data is null for UPDATE/DELETE event. "
+ + "This usually happens when the source table is not
configured with "
+ + "REPLICA IDENTITY FULL. Schema name: "
+ + valueSchema.name());
+ return super.extractBeforeDataRecord(value, valueSchema);
+ }
+
Review Comment:
Could you please add a unit test for `PostgresEventDeserializer`, and an IT
case with a PostgreSQL table without `REPLICA IDENTITY FULL` enabled?
##########
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:
As this PR is PostgreSQL specific, I tend to keep the base class untouched,
and modify 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]
