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

Question

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

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

它提供了我通过Kafka消费者给Kafka经纪人写的所有消息，没有任何遗漏。

最近，我将该应用程序部署在另一个环境中，因为某些原因，zookeperhost无法访问。所以我使用的是kafka简单的消费者外壳。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 读取 myTopic 分区 0 中的一条消息。如果分区 1 不在同一代理中，则它将无法像您所做的那样工作。（有关详细信息，请查看来自 GitHub 的代码）

如果您可以访问Zookeeper主机，那么只需使用

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

但是如果你不能访问动物园管理员的主机，我能想到的有两种方法。

1. 提供一个将所有代理作为参数的列表，并尝试从 0 到 N 的分区号。您可以向 --broker-list 提供多个代理，格式为 broker1：port2，broker2：port2，broker3：port3。然后，您可以计算出整个集群中存在多少个分区，但您仍然不知道哪个代理具有哪些分区。
2. 手动检查每个代理的日志目录。检查 /tmp/kafka-logs（如果您使用的是默认日志目录）。你会发现像myTopic-0，myTopic-1，...采用主题分区#的格式。您可以使用此功能手动检查哪个代理具有哪些分区。