Kafka 简单消费者间歇性丢失消息

Question

我有一个 Kafka 应用程序，我一直在使用 kafka-console-consumer.sh 来消费消息，如下所示：

$./kafka-console-consumer.sh --zookeeper zookeeperhost:2181 --topic myTopic

它提供了我通过 Kafka 消费者写入 Kafka 代理的所有消息，没有任何遗漏。

最近我将应用程序部署在无法访问 zookeeperhost 的不同环境中（由于某种原因）。所以我改用 kafka-simple-consumer-shell.sh 如下：

$./kafka-simple-consumer-shell.sh --broker-list brokerhost:9092 --topic myTopic --partition 0 --max-messages 1

但是，我看到很少有消息（大约 5000 条中的 2-4 条）丢失。有人可以解释一下 kafka-simple-consumer-shell.sh 是如何读取消息的。

我怀疑有些消息可能会发送到某个不同的分区，因为我只是从分区 0 读取，所以我不会每次都收到所有消息。但是我不知道如何检查有多少个分区？其他分区的 id 是什么？我尝试了 1 但它不起作用。

有人可以帮忙吗。

Answer 1

kafka-simple-consumer.sh 只是创建一个从一个分区读取消息的消费者。因此，您的命令只是从brokerhost:9092 读取partition 0 of myTopic 中的一条消息。如果分区 1 不在同一个代理中，它将无法像您所做的那样工作。 （更多信息请查看Code from GitHub）

如果你可以访问 Zookeeper 主机，你可以简单地检查分区在集群中的分布情况

bin/kafka-topics.sh --describe --zookeeper zookeeperhost:2181 --topic myTopic

但是如果不能访问Zookeeper主机，我能想到的有两种方式。

1. 提供一个以所有代理为参数的列表，并尝试从 0 到 N 的分区号​​。您可以以broker1:port2,broker2:port2,broker3:port3 的格式向--broker-list 提供多个代理。然后就可以算出整个集群有多少个partition，但是还是不知道哪个broker有哪个partition。
2. 手动检查每个代理的日志目录。检查/tmp/kafka-logs（如果您使用的是默认日志目录）。您会发现myTopic-0、myTopic-1、...等目录的格式为topic-partition#。您可以使用此手动检查哪个代理具有哪些分区。