laughingman7743 commented on code in PR #28498:
URL: https://github.com/apache/flink/pull/28498#discussion_r4431209368
##########
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:
Thanks for driving the read-as-written approach. It is what makes
`source.properties` carry connector-specific fields, and I would like to use it
for PostgreSQL `source.lsn` and `source.sequence`, which a downstream CDC sink
uses to order changes. With this PR applied to current master, a
PostgreSQL-shaped envelope that requests only `source.properties` returns
`lsn`, `sequence` and `txId` as expected.
One problem remains with this initialization. The generic reader schema is
fixed to the writer schema of the first record this instance sees, and
`RegistryAvroDeserializationSchema` resolves every later record's writer schema
against it. Once the subject gains a new schema version (a table column change,
or a Debezium upgrade that adds `source` fields such as `ts_us`/`ts_ns` in
2.6), later records are no longer read as written:
- A `source` field that only the later version has is dropped from the map,
without an error, until the job restarts.
- A field that is required in the first version and absent from the later
one fails resolution. For example, a NOT NULL column dropped from the database
table but never declared in the Flink DDL gives `AvroTypeException: Found
pg.public.orders.Value, expecting pg.public.orders.Value, missing required
field legacy`, rethrown as `Can't deserialize Debezium Avro message.`
The second case fails only when a metadata column is declared. Without one,
the same two records deserialize, because the typed path resolves each writer
schema against the DDL's reader schema.
I reproduced both by registering two versions under one subject in
`MockSchemaRegistryClient` and passing a v1 record and then a v2 record through
the same deserializer instance (test below).
Reading each message with its own writer schema fixes both cases. With the
change below, the existing `DebeziumAvroSerDeSchemaTest` and
`DebeziumAvroFormatFactoryTest` (12 tests) and the reproduction tests all pass:
```java
private transient SchemaCoder writerCoder;
private transient Map<Schema, GenericDatumReader<GenericRecord>>
writerReaders;
private transient BinaryDecoder writerDecoder;
// in deserialize(), replacing the lazily created genericDeserializer
if (hasMetadata) {
if (writerCoder == null) {
writerCoder = coderProvider.get();
writerReaders = new HashMap<>();
}
ByteArrayInputStream in = new ByteArrayInputStream(message);
Schema writerSchema = writerCoder.readSchema(in);
GenericDatumReader<GenericRecord> reader =
writerReaders.computeIfAbsent(writerSchema,
GenericDatumReader::new);
writerDecoder = DecoderFactory.get().binaryDecoder(in, writerDecoder);
GenericRecord envelope = reader.read(null, writerDecoder);
GenericRecord sourceRecord = (GenericRecord) envelope.get("source");
this.cachedSourceMap = convertSourceToMap(sourceRecord);
}
```
`ConfluentSchemaRegistryCoder` already caches schemas by id, so the added
per-message cost is the five-byte header read and a map lookup.
`genericDeserializer` and `initContext` are then unused.
<details>
<summary>Reproduction test (same package, imports omitted; uses the
@VisibleForTesting constructor as the existing metadata tests do)</summary>
```java
class Flink20454ReproTest {
private static final String SUBJECT = "pg.public.orders-value";
/** Table DDL: id INT NOT NULL, name STRING. */
private static final DataType PHYSICAL =
DataTypes.ROW(
DataTypes.FIELD("id", DataTypes.INT().notNull()),
DataTypes.FIELD("name", DataTypes.STRING()));
private final MockSchemaRegistryClient client = new
MockSchemaRegistryClient();
private final ConfluentSchemaRegistryCoder coder =
new ConfluentSchemaRegistryCoder(SUBJECT, client);
// ----------------------------------------------------------------
schemas
private static Schema value(boolean withLegacyRequiredColumn) {
SchemaBuilder.FieldAssembler<Schema> f =
SchemaBuilder.record("Value")
.namespace("pg.public.orders")
.fields()
.requiredInt("id")
.optionalString("name");
if (withLegacyRequiredColumn) {
f = f.requiredInt("legacy");
}
return f.endRecord();
}
private static Schema source(boolean withTsUs) {
SchemaBuilder.FieldAssembler<Schema> f =
SchemaBuilder.record("Source")
.namespace("io.debezium.connector.postgresql")
.fields()
.requiredString("version")
.requiredString("connector")
.requiredString("name")
.requiredLong("ts_ms")
.optionalString("snapshot")
.requiredString("db")
.optionalString("sequence")
.requiredString("schema")
.requiredString("table")
.optionalLong("txId")
.optionalLong("lsn")
.optionalLong("xmin");
if (withTsUs) {
f = f.optionalLong("ts_us");
}
return f.endRecord();
}
private static Schema envelope(Schema value, Schema source) {
return SchemaBuilder.record("Envelope")
.namespace("pg.public.orders")
.fields()
.name("before").type().optional().type(value)
.name("after").type().optional().type(value)
.name("source").type(source).noDefault()
.requiredString("op")
.optionalLong("ts_ms")
.endRecord();
}
// ----------------------------------------------------------------
messages
private byte[] insert(Schema env, int id, long lsn, boolean legacy,
boolean tsUs)
throws Exception {
Schema valueSchema =
env.getField("after").schema().getTypes().get(1);
Schema sourceSchema = env.getField("source").schema();
GenericRecord after = new GenericData.Record(valueSchema);
after.put("id", id);
after.put("name", "n" + id);
if (legacy) {
after.put("legacy", 7);
}
GenericRecord src = new GenericData.Record(sourceSchema);
src.put("version", "2.7.0.Final");
src.put("connector", "postgresql");
src.put("name", "pg");
src.put("ts_ms", 1_700_000_000_000L);
src.put("snapshot", "false");
src.put("db", "shop");
src.put("sequence", "[\"" + (lsn - 8) + "\",\"" + lsn + "\"]");
src.put("schema", "public");
src.put("table", "orders");
src.put("txId", 4242L);
src.put("lsn", lsn);
if (tsUs) {
src.put("ts_us", 1_700_000_000_000_123L);
}
GenericRecord e = new GenericData.Record(env);
e.put("after", after);
e.put("source", src);
e.put("op", "c");
e.put("ts_ms", 1_700_000_000_500L);
ByteArrayOutputStream out = new ByteArrayOutputStream();
coder.writeSchema(env, out);
BinaryEncoder enc = EncoderFactory.get().binaryEncoder(out, null);
new GenericDatumWriter<GenericRecord>(env).write(e, enc);
enc.flush();
return out.toByteArray();
}
// ----------------------------------------------------------------
harness
private DebeziumAvroDeserializationSchema
deserializer(List<ReadableMetadata> metadata)
throws Exception {
DataType produced =
DataTypeUtils.appendRowFields(
PHYSICAL,
metadata.stream()
.map(m -> DataTypes.FIELD(m.key, m.dataType))
.collect(Collectors.toList()));
RowType rowTypeDe =
DebeziumAvroDeserializationSchema.createDebeziumAvroRowType(PHYSICAL, metadata);
DebeziumAvroDeserializationSchema.MetadataConverter[] converters =
metadata.stream()
.map(
m -> {
int p =
rowTypeDe
.getFieldNames()
.indexOf(m.requiredAvroField.getName());
return
(DebeziumAvroDeserializationSchema.MetadataConverter)
(row, pos) ->
m.converter.convert(row, p);
})
.toArray(DebeziumAvroDeserializationSchema.MetadataConverter[]::new);
// Same reader-schema construction as the production constructor.
AvroRowDataDeserializationSchema typed =
new AvroRowDataDeserializationSchema(
new RegistryAvroDeserializationSchema<>(
GenericRecord.class,
AvroSchemaConverter.convertToSchema(rowTypeDe, false),
() -> coder),
AvroToRowDataConverters.createRowConverter(rowTypeDe, false),
InternalTypeInfo.of(rowTypeDe));
DebeziumAvroDeserializationSchema d =
new DebeziumAvroDeserializationSchema(
InternalTypeInfo.of(produced.getLogicalType()),
typed,
!metadata.isEmpty(),
converters,
rowTypeDe.getFieldNames().indexOf("source"),
() -> coder);
d.open(new Ctx());
return d;
}
private static List<RowData> run(DebeziumAvroDeserializationSchema d,
byte[] msg)
throws Exception {
List<RowData> out = new ArrayList<>();
d.deserialize(
msg,
new Collector<RowData>() {
@Override
public void collect(RowData r) {
out.add(r);
}
@Override
public void close() {}
});
return out;
}
private static String prop(RowData row, int pos, String key) {
MapData m = row.getMap(pos);
StringData v = (StringData) ((GenericMapData)
m).get(StringData.fromString(key));
return v == null ? null : v.toString();
}
// ---------------------------------------------------------------- tests
/** B: a later schema version adds a source field; does the map carry
it? */
@Test
void laterSchemaVersionAddsSourceField() throws Exception {
Schema v1 = envelope(value(false), source(false));
Schema v2 = envelope(value(false), source(true));
client.register(SUBJECT, new AvroSchema(v1));
client.register(SUBJECT, new AvroSchema(v2));
DebeziumAvroDeserializationSchema d =
deserializer(Collections.singletonList(ReadableMetadata.SOURCE_PROPERTIES));
run(d, insert(v1, 1, 1000L, false, false));
List<RowData> rows = run(d, insert(v2, 2, 2000L, false, true));
assertThat(prop(rows.get(0), 2, "lsn")).isEqualTo("2000");
assertThat(prop(rows.get(0), 2,
"ts_us")).isEqualTo("1700000000000123");
}
/** C control: no metadata; v2 drops a required column the DDL never
declared. */
@Test
void droppedUndeclaredRequiredColumnWithoutMetadata() throws Exception {
Schema v1 = envelope(value(true), source(false));
Schema v2 = envelope(value(false), source(false));
client.register(SUBJECT, new AvroSchema(v1));
client.register(SUBJECT, new AvroSchema(v2));
DebeziumAvroDeserializationSchema d =
deserializer(Collections.emptyList());
run(d, insert(v1, 1, 1000L, true, false));
List<RowData> rows = run(d, insert(v2, 2, 2000L, false, false));
assertThat(rows.get(0).getInt(0)).isEqualTo(2);
}
/** C: the same stream with source.properties declared. */
@Test
void droppedUndeclaredRequiredColumnWithMetadata() throws Exception {
Schema v1 = envelope(value(true), source(false));
Schema v2 = envelope(value(false), source(false));
client.register(SUBJECT, new AvroSchema(v1));
client.register(SUBJECT, new AvroSchema(v2));
DebeziumAvroDeserializationSchema d =
deserializer(Collections.singletonList(ReadableMetadata.SOURCE_PROPERTIES));
run(d, insert(v1, 1, 1000L, true, false));
List<RowData> rows = run(d, insert(v2, 2, 2000L, false, false));
assertThat(rows.get(0).getInt(0)).isEqualTo(2);
assertThat(prop(rows.get(0), 2, "lsn")).isEqualTo("2000");
}
private static class Ctx implements
DeserializationSchema.InitializationContext {
@Override
public MetricGroup getMetricGroup() {
return new UnregisteredMetricsGroup();
}
@Override
public UserCodeClassLoader getUserCodeClassLoader() {
return
SimpleUserCodeClassLoader.create(getClass().getClassLoader());
}
}
}
```
</details>
--
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]
