Kafka Streams分组统计结果在Consumer中显示异常求助

你的问题核心在于Kafka Streams输出的消息序列化/反序列化不匹配，导致消费者无法正确解析统计结果。

问题根源 #

1. 在WordCount代码中，count()操作返回的是KTable<String, Long>，转为KStream后value类型是Long。但你全局配置的默认value serde是Serdes.String()，当调用fil.to("temp-out")时，Kafka Streams会尝试用StringSerde序列化Long类型的数据，这会生成错误的字节数据（甚至可能触发隐式类型转换异常）。
2. 消费者代码中，value deserializer用的是StringDeserializer，但主题中存储的是错误序列化的Long数据，所以读取后显示为不可读方框或空字符串。

修复步骤 #

1. 修改WordCount代码，指定正确的Serde

在写入temp-out主题时，显式指定key和value的serde，确保Long类型被正确序列化：

// 替换原来的fil.to("temp-out");
fil.to("temp-out", Produced.with(Serdes.String(), Serdes.Long()));

或者更严谨的方式：在count()操作时就指定存储用的serde，避免后续流操作的类型隐患：

KStream<String, Long> fil = input.flatMapValues(val -> Arrays.asList(val.split(" ")))
        .selectKey((k, v) -> v)
        .groupByKey()
        .count(Materialized.with(Serdes.String(), Serdes.Long())) // 指定count的存储serde
        .toStream();
fil.to("temp-out", Produced.with(Serdes.String(), Serdes.Long()));

2. 修改Consumer代码，匹配反序列化器

将消费者的value deserializer改为LongDeserializer，对应主题中的Long类型数据：

// 替换原来的StringDeserializer配置
properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.LongDeserializer.class.getName());

同时调整消费者的泛型类型，以及循环输出的代码：

// 泛型改为<String, Long>
KafkaConsumer<String, Long> consumer = new KafkaConsumer<>(properties);

// 循环输出时适配Long类型的value
while (true) {
    ConsumerRecords<String, Long> consumerRecords = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, Long> record: consumerRecords) {
        System.out.println(record.key() + " " + record.value());
        System.out.println(record.partition() + " " + record.offset());
    }
}

验证修复 #

1. 停止之前运行的WordCount和Consumer应用。
2. 清理temp-out主题（或重置消费者组偏移量），避免旧的错误数据干扰。
3. 重新启动WordCount应用，发送测试消息到temp-in主题。
4. 启动修改后的Consumer，现在应该能正确看到每个单词的计数结果了。

内容的提问来源于stack exchange，提问作者Bharat