Kafka consumer分区获取最大字节的奇怪行为

Question

我有一个主题a，有12个分区。我在一个集群中有3个Kafka经纪人。对于主题A，每个代理有4个分区。我没有创建任何副本，因为我不关心恢复能力。

我有一个简单的Java消费者使用kafka客户端库。我在属性中提到了以下内容

        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-serverA:9092,kafka-serverB:9092,kafka-serverC:9092");
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupID);
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.setProperty("max.partition.fetch.bytes", "100000");

消费者记录和打印记录的代码更多，工作正常。我在主题中有12条消息，我通过“kafka-run-class.shkafka.admin.消费者组命令”验证了每个分区中都有一条消息。消息大小为100000字节，正好等于max.partition.fetch.bytes限制。

当我轮询时，我应该会看到12条消息作为响应返回。然而，响应非常不稳定。有时我看到来自4个分区的消息，表明只有一个代理正在响应消费者请求，或者有时我看到8个。我从未得到所有12个分区的响应。只是为了测试，我删除了max.partition.fetch.bytes属性。我观察到了同样的行为。

我遗漏了什么吗？引导配置中的serve1、server2、server3似乎在服务请求时没有选择所有3个代理。

非常感谢您的帮助。我在不同的机器上运行经纪人和消费者，他们的规模足够大。

Answer 1

我遗漏了什么吗？引导配置中的serve1、server2、server3似乎在服务请求时没有选择所有3个代理。

在您的Kafkabootstrap.servers属性中，您列出了所有好的代理。其中一个代理将被选中以获取元数据，这基本上是有关主题有多少分区以及哪个代理是这些分区的领导者的信息。

无论选择哪个服务器，都应该提供有关另一个服务器的信息。

检查您的所有代理是否相互认识，即他们属于同一个Kafka群集，即他们指向同一个zookeeper实例。

您提到的所有代理IP必须可供您的消费者访问。因此，请确保已设置适当的播发。侦听器属性。

例如，如果播发。侦听器=纯文本：//1.2.3.4:9092然后消费者必须可以访问1.2.3.4:9092。

此外，默认情况下，Kafka消息会在一段时间后定期自动提交，如果某个消费者使用特定的组ID读取消息，则不会再次使用这些消息，因为它们已提交。因此，您也可以尝试更改组。id属性并重新检查。

此外，请检查是否正在使用相同的组id运行多个使用者，在这种情况下，一些分区将分配给一个使用者，另一些分配给另一个使用者。

您可以通过使用kafka console consumer（Kafka控制台消费者）和给出主题的从头标记来解决此问题，并查看是否所有消息都已被使用。

您可能还需要检查default.api.timeout.ms参数并尝试增加值，以防出现任何网络拥塞导致客户端从一个引导服务器切换到另一个。