Allow kafka sink to encode as PLAIN AVRO about risingwave

[xiangjinwu]

It is doable. Just to confirm:

• Is your avro schema definition stored in confluent schema registry (or one compatible with it, like redpanda or upstash)?
• Are you going to use (a) a subset of output columns as avro-encoded kafka key (b) a single simple output column as string/text kafka key (c) no kafka key and round-robin for partition?

[maingoh]

Thank you for your quick reply ! I might have misunderstood a bit the UPSERT keyword. It didn't really make sense to me as kafka is append only. So I tried using an UPSERT AVRO sink and it actually append the message with the same primary key, which is the behavior I want but It looks more like a PLAIN AVRO to me.

To answer your questions:

• Yes it is the case, however it should not be mandatory in both case PLAIN and UPSERT no ?
• I actually use an "id" column from my materialized view as key. Though it should not be mandatory neither ?

[tabVersion]

I might have misunderstood a bit the UPSERT keyword. It didn't really make sense to me as kafka is append only. So I tried using an UPSERT AVRO sink and it actually append the message with the same primary key, which is the behavior I want but It looks more like a PLAIN AVRO to me.

Yes, it is the tricky part. If the Kafka's key is aligned with the downstream's pk, it is upsert format in RisingWave. If the Kafka's key is not the downstream's pk, just for doing partition, it is append-only format with a specified key in RisingWave. For Kafka itself, the two methods look the same.
Besides, Kafka has a "compaction" func working in the broker, if two messages have the same message key, it will delete the prior only keep the later one. So we want to make sure whether users want to do upsert.

[xiangjinwu]

• Yes it is the case, however it should not be mandatory in both case PLAIN and UPSERT no ?

Use of schema registry is mandatory for avro unless the schema never evolves. Unlike json or protobuf, to decode an avro message encoded with one version of schema into another compatible version, the definitions (avsc) of both versions must be available during decoding.

• I actually use an "id" column from my materialized view as key. Though it should not be mandatory neither ?

It is not mandatory for plain as I mentioned option-c for round-robin partition. It is mandatory for upsert as @tabVersion explained above.