Kafka流-使用Protobuf serde获取问题

Question

我正在创建一个Kafka Streams应用程序，我的主题数据来自Protobuf。我们可以为此创建Java代码绑定。然而，我正在努力使用正确的serde来使用来自主题的数据。有人能告诉我我做错了什么吗。

以下是我使用的属性定义：

Properties properties = new Properties();
    properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app-id-config");
    properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "my-broker:my-port");
    properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, KafkaProtobufSerde.class);

    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

我的Serde课

public class AppSerdes extends Serdes {

    public static KafkaProtobufSerde<ProtobufClass1> createConfiguredSerde1() {
        KafkaProtobufSerde<ProtobufClass1> serde = new KafkaProtobufSerde<ProtobufClass1>();
        Map<String, Object> serdeConfig = getSerdeConfig();
        serde.configure(serdeConfig, false);
        return serde;
    }

    public static KafkaProtobufSerde<ProtobufClass2> createConfiguredSerde2() {
        KafkaProtobufSerde<ProtobufClass2> serde = new KafkaProtobufSerde<ProtobufClass2>();
        Map<String, Object> serdeConfig = getSerdeConfig();
        serde.configure(serdeConfig, false);
        return serde;
    }

    private static Map<String, Object> getSerdeConfig() {
        Map<String, Object> serdeConfig = new HashMap<>();
        serdeConfig.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

        return serdeConfig;
    }
}

这就是我创建KStream和KTable实例的方式：

StreamsBuilder streamBuilder = new StreamsBuilder();
    KTable<String, ProtobufClass1> table = streamBuilder.table("topic1",
            Consumed.with(AppSerdes.String(), AppSerdes.createConfiguredSerde1()));
    KStream<String, ProtobufClass2> stream = streamBuilder.stream("topic2".
            Consumed.with(AppSerdes.String(), AppSerdes.createConfiguredSerde2()));

然而，我得到以下错误：

组织。阿帕奇。Kafka。溪流。错误。StreamsException：调用处理器的ClassCastException。处理器的输入类型是否与反序列化类型匹配？检查Serde设置并更改StreamConfig中的默认Serde，或通过方法参数提供正确的Serde。确保处理器可以接受key:java类型的反序列化输入。字符串和值：com.google.protobuf。动态消息。请注意，尽管错误的SERDE是导致错误的常见原因，但强制转换异常可能还有其他原因（例如，在用户代码中）。例如，如果处理器在存储中连接，但不正确地强制转换泛型，则在处理过程中可能会引发类强制转换异常，但原因不会是错误的Serdes。在org。阿帕奇。Kafka。溪流。加工机内部。处理器节点。进程（ProcessorNode.java:185）位于org。阿帕奇。Kafka。溪流。加工机内部。ProcessorContextImpl。org上的forwardInternal（ProcessorContextImpl.java:273）。阿帕奇。Kafka。溪流。加工机内部。ProcessorContextImpl。在org上转发（ProcessorContextImpl.java:252）。阿帕奇。Kafka。溪流。加工机内部。ProcessorContextImpl。在org上转发（ProcessorContextImpl.java:219）。阿帕奇。Kafka。溪流。加工机内部。SourceNode。在org上处理（SourceNode.java:86）。阿帕奇。Kafka。溪流。加工机内部。简化任务。组织上的lambda$process$1（StreamTask.java:703）。阿帕奇。Kafka。溪流。加工机内部。韵律学。streamsmetricsiml。可以在org上获得测量相关性（streamsmetricsiml.java:883）。阿帕奇。Kafka。溪流。加工机内部。简化任务。流程（StreamTask.java:703）位于组织。阿帕奇。Kafka。溪流。加工机内部。任务经理。org上的进程（TaskManager.java:1105）。阿帕奇。Kafka。溪流。加工机内部。流线型。runOnce（StreamThread.java:647）位于org。阿帕奇。Kafka。溪流。加工机内部。流线型。org上的runLoop（StreamThread.java:553）。阿帕奇。Kafka。溪流。加工机内部。流线型。运行（StreamThread.java:512）的原因是：java。lang.ClassCastException:com.google.protobuf。无法将DynamicMessage强制转换为iit。数据中心。政党系统与客户关系管理。v1。CustomerAddressBase$CustomerAddressBaseEntity位于组织。阿帕奇。Kafka。溪流。kstream。内部。KStreamImpl。org上的lambda$internalSelectKey$0（KStreamImpl.java:234）。阿帕奇。Kafka。溪流。kstream。内部。KStreamMap$kstreamapprocessor。在org上处理（KStreamMap.java:41）。阿帕奇。Kafka。溪流。加工机内部。处理器适配器。进程（ProcessorAdapter.java:71）位于org。阿帕奇。Kafka。溪流。加工机内部。处理器节点。org上的lambda$process$2（ProcessorNode.java:181）。阿帕奇。Kafka。溪流。加工机内部。韵律学。streamsmetricsiml。可以在org上获得测量相关性（streamsmetricsiml.java:883）。阿帕奇。Kafka。溪流。加工机内部。处理器节点。进程（ProcessorNode.java:181）。。。还有11个

Answer 1

我可以通过将此更改为

Map<String, Object> serdeConfig = getSerdeConfig();

到

Map<String, String> serdeConfig = getSerdeConfig();

作为关键