带有 protobuf 特定消息的 Confluent serdes 不起作用答案

【问题标题】：Confluent serdes with protobuf specific messages not working带有 protobuf 特定消息的 Confluent serdes 不起作用
【发布时间】：2022-10-13 02:34:38
【问题描述】：

我很难理解 Confluent serdes apis 的用法。我正在使用带有 protobuf 的 Confluent 平台（7.2.2）模式注册表，并打算在我的 Kafka 流应用程序中使用特定的消息类。

以下是来自here 的示例：

private static KafkaProtobufSerde<Message> createConfiguredSerdeForRecordValues() {
  SchemaRegistryClient schemaRegistryClient = new SchemaRegistryClient();
  KafkaProtobufSerde<Message> serde = new KafkaProtobufSerde<>(schemaRegistryClient);
  Map<String, Object> serdeConfig = new HashMap<>();
  serdeConfig.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "demo");
  serde.configure(serdeConfig, false);
  return serde;
}

在 7.2.2 版本中，SchemaRegistryClient 是抽象的。取而代之的是，我用

var schemaClient = new CachedSchemaRegistryClient(schemaRegUrl, 100);
var assetKeySerde = new KafkaProtobufSerde<>(schemaClient, AssetKey.class);
var assetConfigSerde = new KafkaProtobufSerde<>(schemaClient, AssetConfig.class);

然后最终：

KTable<AssetKey, AssetConfig> assetTable = builder.table(assetTopic, Consumed.with(assetKeySerde, assetConfigSerde));

这里AssetKey 和AssetConfig 是我的protobuf 生成的类。但是，即使在此构造函数中传递 schemaClient 和 protobuf 类，它仍然希望我将映射中的模式注册表 url 和 protobuf 类传递给 .configure()。那么我上面使用的构造函数有什么意义呢？使用此配置，我得到一个错误

com.google.protobuf.DynamicMessage cannot be cast to class AssetKey

如果我在.configure() 中使用schema.registry.url 和specific.protobuf.key.type 传递一个映射，我会收到一个异常，提示架构无效。该架构是有效的，我的制作人可以成功地发布到该主题。

我可以参考一个完整的例子吗？

【问题讨论】：

标签： java apache-kafka protocol-buffers apache-kafka-streams confluent-schema-registry

【解决方案1】：

您是否查看过教程 https://developer.confluent.io/tutorials/changing-serialization-format/kstreams.html 中的示例，源代码位于 https://github.com/confluentinc/kafka-tutorials/blob/master/_includes/tutorials/serialization/kstreams/code/src/main/java/io/confluent/developer/serialization/SerializationTutorial.java ？

我认为将SchemaRegistryClient 传递给KafkaProtobufSerde 的构造函数仅用于测试目的。设置属性就足够了。

【讨论】：

我只使用配置映射就让它工作并忽略了该构造函数。