Hi, I’m trying to create a Flink job using the CDC connectors. It will read multiple MySQL tables in a single source and route each table change to its own topic. However, I need to serialize the data as Avro, and each record might have a different schema, as the stream is processing changes from multiple tables. While trying this, I got a Kryo exception:

Caused by: com.esotericsoftware.kryo.KryoException: java.lang.UnsupportedOperationException
Serialization trace:
reserved (org.apache.avro.Schema$Field)
fieldMap (org.apache.avro.Schema$RecordSchema)
schema (org.apache.avro.generic.GenericData$Record)

It seems like the Kryo serializer cannot handle

GenericRecords

with generic schemas. Currently my code looks like this:

static DebeziumDeserializationSchema<GenericRecord> avroParser = new DebeziumDeserializationSchema<>() {
    private transient AvroData avroData;

    @Override
    public void deserialize(SourceRecord record, Collector<GenericRecord> out) {
        if (avroData == null) {
            avroData = new AvroData(2048);
        }

        var recordValue = record.value();
        var recordSchema = record.valueSchema();
        var avroRecord = (GenericRecord) avroData.fromConnectData(recordSchema, recordValue);

        out.collect(avroRecord);
    }

    @Override
    public TypeInformation<GenericRecord> getProducedType() {
        return TypeInformation.of(GenericRecord.class);
    }
};

// List of tables to consume
var tables = List.of(...);

var mySqlSource = MySqlSource.<GenericRecord>builder()
    .hostname("host")
    .port(3306)
    .username("user")
    .password("password")
    .databaseList("database")
    .tableList(String.join(",", tables))
    .serverTimeZone("UTC")
    .deserializer(avroParser)
    .build();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

var sourceTables = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");

var sink = KafkaSink.<GenericRecord>builder()
    .setBootstrapServers("broker:9092")
    // TODO: Route each message to its corresponding topic based on the table name - record.getSource().getName()
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .build();

sourceTables.sinkTo(sink);

Is there a way to do this?

Might be useful: https://sap1ens.com/blog/2022/07/10/flink-cdc-for-postgres-lessons-learned/

Oh, so you just pass the SourceRecord directly. I though that I would have the same problems with Kryo. I will try this. Thank you!

Could you try

env.getConfig.enableObjectReuse()

?

It worked!

My understanding is that it helps to “fuse” operators together, so the data is passed as is, there is nothing to serialize. Which means Kryo is not used at all.

Makes sense. Thank you

Technically - not really, I think it should work.

Interesting article. And I've been using side-outputs. @sap1ens have you ever faced issues when increasing parallelism using the Table API? maybe written about it somewhere? I have a few "ChangelogNormalize" operators (which I think are created because of the

fromChangelogStream

) and I want to parallelize them as they're the cause of backpressure. However, if I parallelize them, that means that a row with, say, id 0, might be routed to different "ChangelogNormalize" subtasks and create two materialized states for row with id 0. But I guess that can be solved with keyBy, ensuring every row with id 0 goes to the same "ChangelogNormalize" operator. But my #1 worry are JOINs. AFAIK joins can't be parallelized without causing data inconsistency. Is there a way to set parallelization only for the "ChangelogNormalize" operators then?

I only use DataStream API with Flink CDC.

@sap1ens Thanks! Do you happen to know how Flink handles parallelism for joins? If I join tables A and B, and messages for A and B go to their own chain of subtasks, then how are joins ever going to be matched? subtask 0: Source A -> changelognormalize -> join (needs rows from B to match) -> sink subtask 1: Source B -> changelognormalize -> join (needs rows from A to match) -> sink there's also an unanswered stackoverflow question asking the same: https://stackoverflow.com/questions/73063272/how-does-parallelism-works-when-using-flink-sql unfortunately i can't see anywhere discussing this 😕