Kafka控制台使用者在使用--max-messages时提交错误的偏移量

Question

我有一个版本1.1.0中的kafka控制台消费者，我用它从kafka获取消息。当我使用带有option-max-messages的kafka-console-consumer.sh脚本时，它似乎提交了错误的偏移量。

我创建了一个主题和一个消费者小组，并阅读了一些消息：

/kafka_2.11-1.1.0/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.23:9092 --describe --group my-consumer-group
TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
test.offset     1          374             374             0               -               -               -
test.offset     0          0               375             375             -               -               -

/kafka_2.11-1.1.0/bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.23:9092 --topic test.offset --timeout-ms 1000 --max-messages 10 --consumer.config /kafka_2.11-1.1.0/config/consumer.properties
1 var_1
3 var_3
5 var_5
7 var_7
9 var_9
11 var_11
13 var_13
15 var_15
17 var_17
19 var_19
Processed a total of 10 messages

/kafka_2.11-1.1.0/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.23:9092 --describe --group my-consumer-group
Note: This will not show information about old Zookeeper-based consumers.
Consumer group 'my-consumer-group' has no active members.

TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
test.offset     1          374             374             0               -               -               -
test.offset     0          375             375             0               -               -               -

/kafka_2.11-1.1.0/bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.23:9092 --topic test.offset --timeout-ms 1000 --max-messages 10 --consumer.config /kafka_2.11-1.1.0/config/consumer.properties
[2020-02-28 08:27:54,782] ERROR Error processing message, terminating consumer process:  (kafka.tools.ConsoleConsumer$)
kafka.consumer.ConsumerTimeoutException
        at kafka.consumer.NewShinyConsumer.receive(BaseConsumer.scala:98)
        at kafka.tools.ConsoleConsumer$.process(ConsoleConsumer.scala:129)
        at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:84)
        at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:54)
        at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)
Processed a total of 0 messages

Answer 1

这是关于Kafaka消费者的自动提交功能。如本链接所述：

提交偏移量的最简单的方法是允许使用者为您做这件事。如果您配置了enable.auto.commit=true，那么每隔5秒，使用者将提交您的客户机从poll（）接收到的最大偏移量。五秒间隔是默认值，通过设置auto.commit.interval.ms来控制。就像消费者中的其他东西一样，自动提交是由轮询循环驱动的。无论何时轮询，使用者都会检查是否该提交，如果是，它将提交在上次轮询中返回的偏移量。

因此，在您的情况下，当您的消费者轮询时，它接收到最多500条消息（默认值max.poll.records)，并且在5秒后，它提交从上次轮询返回的最大偏移量（在您的情况下为375条），即使您将max-messages指定为10条。

--max-messages：退出前要消耗的最大邮件数。如果不设置，则消耗是连续的。