p-eye commented on code in PR #28498:
URL: https://github.com/apache/flink/pull/28498#discussion_r4431518874
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java:
##########
@@ -133,30 +262,48 @@ public void deserialize(byte[] message,
Collector<RowData> out) throws IOExcepti
return;
}
try {
+ // Initialize generic deserializer with the first record's runtime
writer schema
Review Comment:
@laughingman7743 Thanks for the report and the reproduction. Applied your
suggestion in
[dd0dbea](https://github.com/apache/flink/pull/28498/commits/dd0dbea775478702ed206cb934eabeb4780b7ed0):
each message is now read with its own writer schema. The only difference is
that the coder and reader cache are initialized `open()` rather than lazily.
Added `testSourcePropertiesWithLaterSchemaVersion` and
`testDroppedUndeclaredRequiredColumnWithMetadata`; both fail without the fix. I
left out the no-metadata case since that path is unchanged.
Could you check it again against your PostgreSQL envelope?
--
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]
