Kafka Streams-不可预测的聚合结果

问题描述

我最近正在学习Apache Kafka Streams并玩世界计数示例。下面是我的代码

public class StreamsStarterApp {
public static void main(String[] args) {
    Properties properties = new Properties();
    properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"streams-starter-app");
    properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
    properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,Serdes.String().getClass());
    properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());

    StreamsBuilder streamsBuilder = new StreamsBuilder();

    KStream<String,String> wordCountPipe = streamsBuilder.stream("word-count-in");

    wordCountPipe.filter((key,value) -> StringUtils.isNoneBlank(value))
            .mapValues(value -> value.toLowerCase())
            .flatMapValues(value -> Splitter.on(",").trimResults().split(value))
            .groupBy((key,value)-> value)
            .count(Named.as("count"))
            .toStream()
            .to("word-count-out",Produced.with(Serdes.String(),Serdes.Long()));

    KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(),properties);
    kafkaStreams.start();
    Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
}

}

我有一个很有趣的发现，就是如果我注释'.mapValues（value-> value.toLowerCase（））'，结果将有所不同，这使我真的感到困惑，因为代码中的任何更改都会导致不可预测的结果更改>

1. 向主题“ word-count-int”发送问候，您好

2. 结果将显示 你好2

3. 如果我评论'.mapValues（value-> value.toLowerCase（））'并再次发送hello，world 结果将显示 你好1 世界1

这怎么可能发生？与卡夫卡流中的状态存储有关吗？

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）