将消费者偏移重置到 Kafka Streams 的开头

Question

我正在使用 Kafka 流，并且想要将一些消费者偏移量从 Java 重置到开头。

KafkaConsumer.seekToBeginning(...)

听起来是正确的做法，但我使用 Kafka Streams：

KafkaStreams streams = new KafkaStreams(builder, props);
...
streams.start();

我猜想根据我定义的具体流管道，这会在幕后创建多个消费者。我可以访问这些吗？或者还有其他方法可以通过编程重置偏移量吗？

Answer 1

基于 Hans Jespersens 的回答，我成功地使用此代码完成了脚本在 Java 代码中执行的操作：

import kafka.tools.StreamsResetter;

StreamsResetter resetter = new StreamsResetter();
String[] args = {"--application-id", APP_ID, "--bootstrap-servers", KAFKA_SERVERS, "--input-topics", TEST_TOPIC_NAME, "--zookeeper", ZOOKEEPER};
resetter.run(args);

该类是我使用 Maven 导入的 kafka 核心库的一部分：

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.12</artifactId>
        <version>${kafka.version}</version>
    </dependency>

Answer 2

由于您使用的是 Kafka Streams，您不仅需要重置消费者偏移量，还需要重置 Streams 内部状态存储。

幸运的是，Kafka 提供了 Streams 应用程序重置工具。

参见 https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool

Answer 3

在 Kafka 流中，有一个技巧 - 如果这适合您，您可以为 APPLICATION_ID_CONFIG 分配一个随机值。然后你总是会从头开始阅读新内容，但这是一种肮脏的方式。