p-eye commented on code in PR #28498:
URL: https://github.com/apache/flink/pull/28498#discussion_r3563432875
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java:
##########
@@ -112,11 +202,50 @@ public DebeziumAvroDeserializationSchema(
AvroRowDataDeserializationSchema avroDeserializer) {
this.producedTypeInfo = producedTypeInfo;
this.avroDeserializer = avroDeserializer;
+ this.hasMetadata = false;
+ this.metadataConverters = new MetadataConverter[0];
+ this.sourceFieldPosition = -1;
+ this.envelopeSchema = null;
+ this.schemaRegistryUrl = null;
+ this.registryConfigs = null;
+ }
+
+ @VisibleForTesting
+ DebeziumAvroDeserializationSchema(
+ TypeInformation<RowData> producedTypeInfo,
+ AvroRowDataDeserializationSchema avroDeserializer,
+ boolean hasMetadata,
+ MetadataConverter[] metadataConverters,
+ int sourceFieldPosition,
+ Schema envelopeSchema,
+ SchemaCoderProvider coderProvider) {
+ this.producedTypeInfo = producedTypeInfo;
+ this.avroDeserializer = avroDeserializer;
+ this.hasMetadata = hasMetadata;
+ this.metadataConverters = metadataConverters;
+ this.sourceFieldPosition = sourceFieldPosition;
+ this.envelopeSchema = envelopeSchema;
+ this.coderProvider = coderProvider;
+ this.genericDeserializer = null;
+ this.schemaRegistryUrl = null;
+ this.registryConfigs = null;
}
@Override
public void open(InitializationContext context) throws Exception {
avroDeserializer.open(context);
+
+ if (hasMetadata && this.genericDeserializer == null) {
+ if (this.coderProvider != null) {
+ this.genericDeserializer =
+ new RegistryAvroDeserializationSchema<>(
+ GenericRecord.class, this.envelopeSchema,
this.coderProvider);
+ } else {
+ this.genericDeserializer =
+ ConfluentRegistryAvroDeserializationSchema.forGeneric(
+ this.envelopeSchema, schemaRegistryUrl,
registryConfigs);
Review Comment:
You're right about the schema projection issue. Thanks for the guidance.
- **Fix**: I’ve switched the generic deserializer to perform lazy
initialization using the actual writer schema from the first record. This
"read-as-written" approach effectively bypasses the empty reader schema
projection and restores full source metadata access.
- **Test Gap**: I’ve updated the tests to use the production
`createRuntimeDecoder` path. This ensures that the assertions now validate the
actual runtime schema resolution, closing the gap you pointed out.
--
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]
