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

如何解决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

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

解决方法

修改KafkaStreams应用程序（即删除或添加运算符）可能会导致不兼容。通常，如果您想更改程序，则通常需要重置应用程序（即，删除所有状态）（参见https://docs.confluent.io/current/streams/developer-guide/app-reset-tool.html）。

对于您的特殊情况，问题在于操作员名称。名称是使用内部计数器自动生成的，以避免命名冲突。如果删除一个运算符，则下游运算符的名称会更改。因此，count()运算符找不到旧状态（每个状态存储区也都有一个名称，并且存储区的名称也发生了变化），因此，在删除mapValues之后，您将以空状态开始

您可以通过Topology#describe()检查命名。这样，您可以在更改代码之前和之后比较拓扑。

要允许兼容升级，DSL允许您显式指定名称（参见https://docs.confluent.io/current/streams/developer-guide/dsl-topology-naming.html）。这样，命名不会更改。对于单词计数示例，您可以通过以下方式指定名称：

.count(Materialized.as("myName"))