weiqingy commented on code in PR #28498:
URL: https://github.com/apache/flink/pull/28498#discussion_r3568435092
##########
flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroSerDeSchemaTest.java:
##########
@@ -146,10 +155,139 @@ void testDeleteDataDeserialization() throws Exception {
assertThat(actual).isEqualTo(expected);
}
+ @Test
+ void testTombstoneMessages() throws Exception {
+ RowType rowTypeDe =
+ DebeziumAvroDeserializationSchema.createDebeziumAvroRowType(
+ fromLogicalToDataType(rowType),
Collections.emptyList());
+ client.register(SUBJECT, DEBEZIUM_SCHEMA_COMPATIBLE_TEST, 1, 81);
+
+ DebeziumAvroDeserializationSchema dbzDeserializer =
+ new DebeziumAvroDeserializationSchema(
+ InternalTypeInfo.of(rowType),
getDeserializationSchema(rowTypeDe));
+ dbzDeserializer.open(new MockInitializationContext());
+
+ SimpleCollector collector = new SimpleCollector();
+ dbzDeserializer.deserialize(null, collector);
+ dbzDeserializer.deserialize(new byte[] {}, collector);
+ assertThat(collector.list).isEmpty();
+ }
+
+ @Test
+ void testMapHelperMethods() {
Review Comment:
This one asserts against `getMapValue`, the test file's own private helper
(`:417`), so it passes or fails independently of anything in the deserializer —
the production `convertSourceToMap` path is already covered by the `sourceMap`
assertions in `testDeserializationWithMetadata`. Is it worth keeping, or would
dropping it shrink the surface to maintain? Non-blocking either way.
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java:
##########
@@ -83,27 +99,104 @@ public final class DebeziumAvroDeserializationSchema
implements DeserializationS
/** TypeInformation of the produced {@link RowData}. */
private final TypeInformation<RowData> producedTypeInfo;
+ /** Flag that indicates that an additional projection is required for
metadata. */
+ private final boolean hasMetadata;
+
+ /** Metadata to be extracted for every record. */
+ private final MetadataConverter[] metadataConverters;
+
+ /** Schema registry URL for generic deserializer. */
+ private final String schemaRegistryUrl;
+
+ /** Schema registry configs for generic deserializer. */
+ private final Map<String, ?> registryConfigs;
+
+ /** Position of source field in rootRow, -1 if not present. */
+ private final int sourceFieldPosition;
+
+ /** Generic Avro deserializer for extracting envelope with writer schema.
*/
+ private transient RegistryAvroDeserializationSchema<GenericRecord>
genericDeserializer;
+
+ /** Initialization context for the generic deserializer. */
+ private transient InitializationContext initContext;
+
+ /** Cached source MapData for current message. */
+ private transient MapData cachedSourceMap;
+
+ /** Provider for mock schema registry in tests. */
+ private transient SchemaCoderProvider coderProvider;
+
+ /**
+ * Converts Debezium source GenericRecord to {@code MAP<STRING, STRING>}.
+ *
+ * <p>Extracts all fields from writer's source schema dynamically,
preserving connector-specific
+ * fields before projection.
+ *
+ * @param sourceRecord GenericRecord from writer schema
+ * @return MapData with all source fields as strings, null if input is null
+ */
+ private static MapData convertSourceToMap(@Nullable GenericRecord
sourceRecord) {
+ if (sourceRecord == null) {
+ return null;
+ }
+
+ Map<StringData, StringData> map = new HashMap<>();
+ Schema sourceSchema = sourceRecord.getSchema();
+
+ for (Schema.Field field : sourceSchema.getFields()) {
+ String fieldName = field.name();
+ Object fieldValue = sourceRecord.get(fieldName);
+ map.put(
+ StringData.fromString(fieldName),
+ fieldValue != null ?
StringData.fromString(fieldValue.toString()) : null);
+ }
+
+ return new GenericMapData(map);
+ }
+
public DebeziumAvroDeserializationSchema(
- RowType rowType,
+ DataType physicalDataType,
+ List<ReadableMetadata> requestedMetadata,
TypeInformation<RowData> producedTypeInfo,
String schemaRegistryUrl,
@Nullable String schemaString,
@Nullable Map<String, ?> registryConfigs) {
this.producedTypeInfo = producedTypeInfo;
- RowType debeziumAvroRowType =
createDebeziumAvroRowType(fromLogicalToDataType(rowType));
+ this.schemaRegistryUrl = schemaRegistryUrl;
+ this.registryConfigs = registryConfigs;
+
+ RowType debeziumAvroRowType =
+ createDebeziumAvroRowType(physicalDataType, requestedMetadata);
validateSchemaString(schemaString, debeziumAvroRowType);
Schema schema =
schemaString == null
- ?
AvroSchemaConverter.convertToSchema(debeziumAvroRowType)
+ ?
AvroSchemaConverter.convertToSchema(debeziumAvroRowType, false)
: new Parser().parse(schemaString);
-
this.avroDeserializer =
new AvroRowDataDeserializationSchema(
ConfluentRegistryAvroDeserializationSchema.forGeneric(
schema, schemaRegistryUrl, registryConfigs),
-
AvroToRowDataConverters.createRowConverter(debeziumAvroRowType),
+
AvroToRowDataConverters.createRowConverter(debeziumAvroRowType, false),
producedTypeInfo);
+ this.hasMetadata = !requestedMetadata.isEmpty();
+ this.sourceFieldPosition =
debeziumAvroRowType.getFieldNames().indexOf("source");
+ this.metadataConverters =
+ requestedMetadata.stream()
+ .map(
+ m -> {
+ final int rootPosition =
+ debeziumAvroRowType
+ .getFieldNames()
+
.indexOf(m.requiredAvroField.getName());
+ return (MetadataConverter)
+ (row, pos) -> {
+ // Simplified: always pass
(rootRow, rootPosition)
Review Comment:
Small thing while you're here: `scala` should be `scalar`, and the
`Simplified:` prefix reads as a note about how the code got here rather than
what it does. Something like this, if useful:
```java
// Converters take (rootRow, rootPosition); each handles its own type —
// scalar for ts_ms, MapData for source.
```
Non-blocking.
--
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]
