weiqingy commented on code in PR #28498:
URL: https://github.com/apache/flink/pull/28498#discussion_r3555144540
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java:
##########
@@ -112,11 +191,37 @@ public DebeziumAvroDeserializationSchema(
AvroRowDataDeserializationSchema avroDeserializer) {
this.producedTypeInfo = producedTypeInfo;
this.avroDeserializer = avroDeserializer;
+ this.hasMetadata = false;
+ this.metadataConverters = new MetadataConverter[0];
+ this.schemaRegistryUrl = null;
+ this.registryConfigs = null;
+ }
+
+ @VisibleForTesting
+ DebeziumAvroDeserializationSchema(
+ TypeInformation<RowData> producedTypeInfo,
+ AvroRowDataDeserializationSchema avroDeserializer,
+ boolean hasMetadata,
+ MetadataConverter[] metadataConverters,
+ @Nullable RegistryAvroDeserializationSchema<GenericRecord>
genericDeserializer) {
+ this.producedTypeInfo = producedTypeInfo;
+ this.avroDeserializer = avroDeserializer;
+ this.hasMetadata = hasMetadata;
+ this.metadataConverters = metadataConverters;
+ this.genericDeserializer = genericDeserializer;
+ this.schemaRegistryUrl = null;
+ this.registryConfigs = null;
}
@Override
public void open(InitializationContext context) throws Exception {
avroDeserializer.open(context);
+
+ if (hasMetadata && this.genericDeserializer == null) {
+ this.genericDeserializer =
+ ConfluentRegistryAvroDeserializationSchema.forGeneric(
+ null, schemaRegistryUrl, registryConfigs);
Review Comment:
The reader schema passed here is `null`. `forGeneric(null, …)` lands in
`AvroDeserializationSchema` with `schemaString == null`, so on the first record
`checkAvroInitialized()` runs `new Schema.Parser().parse(null)`
(`AvroDeserializationSchema.java:200-201`) — which NPEs (`Cannot invoke
"String.length()" because "content" is null`, reproduced against the repo's
Avro 1.11.5), and the outer `catch (Throwable)` at `:279-282` rewraps it as
`Can't deserialize Debezium Avro message.`. This branch runs whenever a
metadata column is declared (`hasMetadata == true`, wired from
`createRuntimeDecoder` at `DebeziumAvroDecodingFormat.java:96-102`), so every
metadata-enabled table would die on its first record. The test doesn't see it
because it injects a `genericDeserializer` built with a real reader schema
instead of going through `open()`. Would passing a concrete reader schema here
— the envelope schema, or one built from the registry — be the intended wiring?
(The missing `.open(co
ntext)` on `genericDeserializer` is harmless on its own; lazy init covers it.
It's the null schema that's fatal.)
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java:
##########
@@ -169,6 +282,45 @@ public void deserialize(byte[] message, Collector<RowData>
out) throws IOExcepti
}
}
+ private void emitRow(
+ GenericRowData rootRow, GenericRowData physicalRow,
Collector<RowData> out) {
+ // shortcut in case no output projection is required
+ if (!hasMetadata) {
+ out.collect(physicalRow);
+ return;
+ }
+
+ // Inject MapData at source position (4) in rootRow
+ final int sourcePosition = 4; // Stable Debezium envelope position
Review Comment:
The root row is `[before(0), after(1), op(2)]` plus the distinct required
Avro fields of only the *requested* metadata, in request order
(`createDebeziumAvroRowType`, `:353-370`). So `source` sits at index 4 only
when `ingestion-timestamp` (contributing `ts_ms`) is also requested and
precedes any `source.*`. The metadata converters already read at the real index
via `debeziumAvroRowType.getFieldNames().indexOf(...)` (`:174-177`); only this
injection is hardcoded, so the two disagree for most selections.
Concretely, request `source.database` alone: root is `[before, after, op,
source]`, `source` at index 3, arity 4 — the loop's `i == 4` guard never fires,
so the injection is skipped, position 3 keeps the empty typed `GenericRowData`,
and `readProperty` → `row.getMap(3)` casts a `GenericRowData` to `MapData`
(`GenericRowData.java:201-203`) → `ClassCastException`. Declaring `source.*`
before `ingestion-timestamp` is worse: the map is injected over the `ts_ms`
slot. The metadata test passes only because it requests all six in enum order,
the one layout where `4` holds.
One option, since the value is already computed for `rootPosition`: derive
it the same way — `debeziumAvroRowType.getFieldNames().indexOf("source")` — and
inject only when a `source` field is present. Happy to be redirected if there's
a reason to keep the literal.
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroSerializationSchema.java:
##########
@@ -81,15 +81,15 @@ public DebeziumAvroSerializationSchema(
validateSchemaString(schemaString, debeziumAvroRowType);
Schema schema =
schemaString == null
- ?
AvroSchemaConverter.convertToSchema(debeziumAvroRowType)
+ ?
AvroSchemaConverter.convertToSchema(debeziumAvroRowType, false)
Review Comment:
This (and the matching `createConverter(..., false)` at `:92`) switches the
serialization schema from the 1-arg default `legacyTimestampMapping = true` to
`false`. The serialization row type is `ROW(before, after, op)` over the user's
physical columns, so for any physical `TIMESTAMP(3)` sink column this changes
the emitted Avro logical type from `timestamp-millis` to
`local-timestamp-millis` (`AvroSchemaConverter.java:462-509`) — a registry/wire
change for existing `debezium-avro-confluent` sinks that have a timestamp
column, in a PR scoped to adding read metadata. `TIMESTAMP_LTZ` is a separate
story: it used to throw, so enabling it here is a real improvement, not a
break. Was the write path meant to change at all, or can it stay on the 1-arg
default? If the change is intended, it likely wants a compatibility note.
##########
flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroSerDeSchemaTest.java:
##########
@@ -164,6 +263,50 @@ 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 {
+ final List<ReadableMetadata> requestedMetadata =
Arrays.asList(ReadableMetadata.values());
+
+ 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);
+
+ RegistryAvroDeserializationSchema<GenericRecord> genericDeserializer =
Review Comment:
Injecting a pre-built, pre-opened `genericDeserializer` here (through the
new 5-arg `@VisibleForTesting` constructor) means the test never runs the
production `open()`, which is where the generic deserializer gets built with a
null reader schema. And `requestedMetadata` above is
`Arrays.asList(ReadableMetadata.values())` (`:268`) — all six in enum order —
which happens to be the one layout where the hardcoded `sourcePosition = 4` in
`emitRow` is correct. Between them, the two production bugs I've flagged on
`DebeziumAvroDeserializationSchema` both sit in the gap between this test and
`createRuntimeDecoder`, which is why they slip through green. A case that
drives the real `open()` path, plus one that requests `source.database` alone
without `ingestion-timestamp`, would each catch one of them. Worth adding both?
--
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]
