apache-kafka - 在 Kafka-python 中的消费者组中重置 kafka LAG(更改偏移量)

Question

我在使用 kafka-consumer-groups.sh 工具 How to change start offset for topic? 重置 LAG 的地方找到了这个但我需要在应用程序中重置它。我找到了这个例子，但它似乎没有重置它。 kafka-python read from last produced message after a consumer restart例子

    consumer = KafkaConsumer("MyTopic", bootstrap_servers=self.kafka_server + ":" + str(self.kafka_port),
                             enable_auto_commit=False,
                             group_id="MyTopic.group")
    consumer.poll()
    consumer.seek_to_end()
    consumer.commit()

    ... continue on with other code...

运行 bin\windows\kafka-consumer-groups.bat --bootstrap-server localhost:9092 --group MyTopic.group --describe仍然显示两个分区都有一个 LAG。我怎样才能让当前偏移“快进”到最后？

TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                             HOST             CLIENT-ID
MyTopic         0          52110           66195           14085           kafka-python-1.4.2-6afb6901-c651-4534-a482-15358db42c22 /Host1  kafka-python-1.4.2
MyTopic         1          52297           66565           14268           kafka-python-1.4.2-c70e0a71-7d61-46a1-97bc-aa2726a8109b /Host2  kafka-python-1.4.2

Answer 1

为了“快进”消费者组的偏移量，意味着清除LAG，您需要创建将加入同一组的新消费者。
控制台命令是:

kafka-console-consumer.sh --bootstrap-server <brokerIP>:9092 --topic <topicName> --consumer-property group.id=<groupName>

同时，您可以运行该命令以查看您描述的延迟，您将看到延迟已被消除。