为Kafka流配置Serdes时出现问题

Question

我在“提交”主题中放置了一个json对象。我想使用Kafka流使用消息，但出现了一个错误

@Configuration
@EnableKafka
@EnableKafkaStreams
public class AnalyzerConfiguration {
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-streams");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, commitSerde().getClass());

        return new KafkaStreamsConfiguration(props);
    }

    @Bean
    public Serde<Commit> commitSerde() {
        return Serdes.serdeFrom(new JsonSerializer<>(), new JsonDeserializer<>(Commit.class));
    }

    @Bean
    public KStream<String, Commit> kStream(StreamsBuilder builder) {
        KStream<String, Commit> stream = builder.stream("commits", Consumed.with(Serdes.String(), commitSerde()));

        KTable<String, Long> commitsCount = stream
                .mapValues(Commit::getAuthorName)
                .selectKey((key, word) -> word)
                .groupByKey()
                .count(Materialized.as("Counts"));

        commitsCount.toStream().to("commits-count", Produced.with(Serdes.String(), Serdes.Long()));

        return stream;
    }
}

日志显示：

线程“test-streams-469f5ee6-d0de-472e-a602-a7b6d11f2e1c-StreamThread-1”组织中出现异常。阿帕奇。Kafka。流。错误。StreamsException：无法配置值serde类组织。阿帕奇。Kafka。常见的序列化。Serdes$包装器Serde

原因：组织。阿帕奇。Kafka。常见的KafkaException：找不到组织的公共无参数构造函数。阿帕奇。Kafka。常见的序列化。Serdes$包装器Serde

原因：java。lang.NoSuchMethodException

Answer 1

您的问题是StreamsConfig的注册。默认值\u SERDE\u CLASS\u配置。首先，在您的示例中不需要这样做，因为您在创建KStream时使用的消费代码中指定了值serde。您可以省略默认的serde。

如果将一个类注册为默认serde，Kafka Streams将在某个时候通过反射创建该类的实例。调用该类的默认（无参数）构造函数。在您的示例中，是一个组织。阿帕奇。Kafka。常见的序列化。Serdes$WrapperSerde类将从Serdes使用。serdeFrom（新型JsonSerializer

如果要为提交类型注册默认serde，需要将其包装到一个小类中：

public class CommitSerde extends WrapperSerde<Commit> {

    public CommitSerde() {
        super(new JsonSerializer<>(), new JsonDeserializer<>(Commit.class));
    }
}

这个类应该适合使用props注册为默认值serde。put（StreamsConfig.DEFAULT\u VALUE\u SERDE\u CLASS\u CONFIG，CommitSerde.CLASS.getName（）） 在您的示例中。