weiqingy commented on code in PR #28498:
URL: https://github.com/apache/flink/pull/28498#discussion_r3563180985
##########
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:
This clears the `parse(null)` NPE — `envelopeSchema` is non-null now — but
it also changes what the generic deserializer reads back. `envelopeSchema` is
built from `debeziumAvroRowType` (`:171`), whose `source` field is
`DataTypes.FIELD("source", DataTypes.ROW())` — an empty Avro record
(`DebeziumAvroDecodingFormat.java:139-140`).
`RegistryAvroDeserializationSchema.deserialize` does `setSchema(writerSchema);
setExpected(readerSchema)` (`RegistryAvroDeserializationSchema.java:102-103`),
so Avro resolves the writer's full `source` record down to that empty reader
record and every writer field is projected away. `convertSourceToMap` then
iterates zero fields (`:141`) and returns an empty map, so on the default path
(no explicit `avro-confluent.schema`) `source.database` / `source.table` /
`source.schema` / `source.timestamp` all come back null at runtime, and
`source.properties` comes back as an empty map (`ingestion-timestamp` is
unaffected — it reads the top-level `ts_ms`).
I checked this directly against Avro 1.11.5: a writer `source{db, table}`
resolved against an empty reader `source{}` yields `{}` — the fields are gone.
The earlier `forGeneric(null, …)` was aiming at "read as written" (writer
schema used as the reader) so the connector-specific fields survived — that's
what made the projected MAP connector-agnostic in the first place. The
empty-`source` reader schema undoes that. Is there a way to keep the generic
read resolving to the writer's `source` while still handing
`AvroDeserializationSchema` a non-null schema so it doesn't NPE on init?
##########
flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroSerDeSchemaTest.java:
##########
@@ -164,6 +302,56 @@ public List<String> testDeserialization(String dataPath)
throws Exception {
return
collector.list.stream().map(Object::toString).collect(Collectors.toList());
}
+ private void testDeserializationWithMetadata(String dataPath,
Consumer<RowData> testConsumer)
+ throws Exception {
+ testDeserializationWithMetadata(
+ dataPath, Arrays.asList(ReadableMetadata.values()),
testConsumer);
+ }
+
+ private void testDeserializationWithMetadata(
+ String dataPath,
+ List<ReadableMetadata> requestedMetadata,
+ Consumer<RowData> testConsumer)
+ throws Exception {
+ final DataType producedDataType =
+ DataTypeUtils.appendRowFields(
+ fromLogicalToDataType(rowType),
+ requestedMetadata.stream()
+ .map(m -> DataTypes.FIELD(m.key, m.dataType))
+ .collect(Collectors.toList()));
+
+ RowType rowTypeDe =
+ DebeziumAvroDeserializationSchema.createDebeziumAvroRowType(
+ fromLogicalToDataType(rowType), requestedMetadata);
+
+ client.register(SUBJECT, DEBEZIUM_SCHEMA_COMPATIBLE_TEST, 1, 81);
+
+ ConfluentSchemaRegistryCoder registryCoder =
+ new ConfluentSchemaRegistryCoder(SUBJECT, client);
+
+ DebeziumAvroDeserializationSchema.MetadataConverter[]
metadataConverters =
+ createMetadataConverters(rowTypeDe, requestedMetadata);
+
+ int sourceFieldPosition = rowTypeDe.getFieldNames().indexOf("source");
+
+ DebeziumAvroDeserializationSchema dbzDeserializer =
+ new DebeziumAvroDeserializationSchema(
+ InternalTypeInfo.of(producedDataType.getLogicalType()),
+ getDeserializationSchema(rowTypeDe),
+ true,
+ metadataConverters,
+ sourceFieldPosition,
+ DEBEZIUM_SCHEMA_COMPATIBLE_TEST,
Review Comment:
This is why the three new tests stay green even though production returns
null source metadata: the test passes `DEBEZIUM_SCHEMA_COMPATIBLE_TEST` — a
full 14-field `source` record — as the `envelopeSchema`, whereas the production
`createRuntimeDecoder` path builds the reduced empty-`source` schema through
`createDebeziumAvroRowType`. So the resolve step that drops the fields never
runs here, and the assertions pass against a schema the runtime never sees.
Would it be worth driving a case through `createRuntimeDecoder` (or building
the reader schema the way production does) so the assertions exercise the
actual runtime schema? That would close the remaining gap in what the test
reaches.
--
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]
