AVRO 原始类型的 Serde 类

我正在用 Java 编写一个 Kafka 流应用程序，它接受由连接器创建的输入主题，该连接器使用架构注册表和 avro 作为键和值转换器。连接器生成以下模式：

key-schema: "int"
value-schema:{
"type": "record",
"name": "User",
"fields": [
    {"name": "firstname", "type": "string"},
    {"name": "lastname",  "type": "string"}
]}

实际上，有几个主题，键模式始终是“int”，值模式始终是某种记录（用户、产品等）。我的代码包含以下定义

Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url", schemaRegistryUrl);

Serde<User> userSerde = new SpecificAvroSerde<>();
userSerde.configure(serdeConfig, false);

起初我尝试用类似的东西来消费这个话题Consumed.with(Serdes.Integer(), userSerde);但这不起作用，因为 Serdes.Integer() 期望使用 4 个字节对整数进行编码，但 avro 使用可变长度编码。使用Consumed.with(Serdes.Bytes(), userSerde);有效，但我真的想要 int 而不是 bytes，所以我将代码更改为这个

KafkaAvroDeserializer keyDeserializer = new KafkaAvroDeserializer()
KafkaAvroSerializer keySerializer = new KafkaAvroSerializer();
keyDeserializer.configure(serdeConfig, true); 
keySerializer.configure(serdeConfig, true);
Serde<Integer> keySerde = (Serde<Integer>)(Serde)Serdes.serdeFrom(keySerializer, keyDeserializer);

这使得编译器产生警告（它不喜欢(Serde<Integer>)(Serde)铸造）但它允许我使用

Consumed.with(keySerde, userSerde);并获取一个整数作为密钥。这工作得很好，我的应用程序的行为符合预期（太棒了！！！）。但现在我想为键/值定义默认的 serde，但我无法让它工作。

设置默认值 serde 很简单：

streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);

但是我不知道如何定义默认密钥 serde。

I tried

1. streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, keySerde.getClass().getName());产生运行时错误：无法找到 org.apache.kafka.common.serialization.Serdes$WrapperSerde 的公共无参构造函数
2. streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);产生运行时错误：java.lang.Integer 无法转换为 org.apache.avro.specific.SpecificRecord

我缺少什么？ 谢谢。

Update （5.5 版及更新版本）

融合版本5.5添加对原始 Avro 类型的本机支持PrimitiveAvroSerde (cf. https://github.com/confluenceinc/schema-registry/blob/5.5.x/avro-serde/src/main/java/io/confluence/kafka/streams/serdes/avro/PrimitiveAvroSerde.java https://github.com/confluentinc/schema-registry/blob/5.5.x/avro-serde/src/main/java/io/confluent/kafka/streams/serdes/avro/PrimitiveAvroSerde.java)

原答案 （5.4 及更早版本）:

这是一个已知问题。原始 Avro 类型不能与 Confluence 的 AvroSerdes 很好地配合使用，因为 Serdes 可以与GenericAvroRecord and SpecificAvroRecord only.

Compare https://github.com/confluenceinc/schema-registry/tree/master/avro-serde/src/main/java/io/confluence/kafka/streams/serdes/avro https://github.com/confluentinc/schema-registry/tree/master/avro-serde/src/main/java/io/confluent/kafka/streams/serdes/avro.

因此，基于以下内容构建您自己的 SerdeKafkaAvroSerializer and KafkaAvroDeserializer是正确的做法。为了能够将其作为默认 Serde 传递到配置中，您不能使用Serdes.serdeFrom因为类型信息由于泛型类型擦除而丢失。

但是，您可以实现自己的扩展类Serde接口，并将您的自定义类传递到配置中：

public class MySerde extends Serde<Integer> {
    // use KafkaAvroSerializer and KafkaAvroDeserializer and cast `Object` to `Integer`
}

config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, MySerde.class);