在Kafka Streams应用程序中，第二个输出流不再被写入

Question

我目前正在实现一个Kafka Streams应用程序，其中我正在阅读一个主题并进行一些处理。在处理过程中，我将其分为两个流，一个写入一个主题（Avro模式），另一个是计数聚合（字数计数），将键/值对（字符串/长）写入另一个主题。代码之前运行良好，但最近第二个流不再编写。

在此代码示例中：

KStream<String, ProcessedSentence> sentenceKStream = stream
        .map((k,v) -> {
                [...]
        });

// configure serializers for publishing to topic
final Serde<ProcessedSentence> valueProcessedSentence = new SpecificAvroSerde<>();
valueProcessedSentence.configure(serdeConfig, false);
stringSerde.configure(serdeConfig, true);

// write to Specific Avro Record
sentenceKStream
        .to(EnvReader.KAFKA_SENTENCES_TOPIC_NAME_OUT,
                Produced.with(
                        Serdes.String(),
                        valueProcessedSentence));

句子流（sentenceKStream）编写正确，但单词计数分组出现了问题：

KStream<String, Long> wordCountKStream =
        sentenceKStream.flatMap((key, processedSentence) -> {
            List<KeyValue<String, Long>> result = new LinkedList<>();
            Map<CharSequence, Long> words = processedSentence.getWords();
            for (CharSequence word: words.keySet() ) {
                result.add(KeyValue.pair(word.toString(), words.get(word)));
            }
            return result;
        })
        .groupByKey(Grouped.with(Serdes.String(), Serdes.Long()))
        .reduce(Long::sum)
        .toStream();

// write to Specific Avro Record
wordCountKStream
        .to(EnvReader.KAFKA_WORDS_COUNT_TOPIC_NAME_OUT,
                Produced.with(
                        Serdes.String(),
                        Serdes.Long()));

我真的不明白为什么不再写wordCountKStream了。

也许有人能提供一些帮助？我很乐意提供更多细节！

很多谢谢

更新：我发现两个新输出流中的数据都丢失了。实际上，一切都写得很正确，但是在写入数据后几分钟，两个主题都被删除了（还剩0字节）。

Answer 1

这与实现本身无关。我只是用

kafka-consumer-groups.sh --bootstrap-server [broker:port] --delete-offsets --group [group_name] --topic [topic_name]

解决了问题。在调试过程中，存储的偏移量出现了问题，并且与streams应用程序的多次重启相冲突。

对于那些想要列出组以查找存储的主题位置的人，请拨打电话

kafka-consumer-groups.sh --bootstrap-server node1:9092 --list

更新：不幸的是，删除组偏移量也没有正常工作。实际的问题是，输出主题中的新条目所用的时间戳是来自原始主题（已消费）的时间戳，它根本没有改变。因此，新条目的时间戳比默认保留时间早。

因为这个话题有一个主题。ms of-1（永久保留数据）和新主题（我认为是6天的标准），消耗的主题中的条目仍然存在，但生成的主题中的条目总是被删除，因为这些条目已超过6天。

最终的解决方案是改变保留率。输出主题从ms到-1。

提示：对于Streams应用程序，建议使用应用程序重置工具，而不是如上所示手动重置/删除偏移。