使用 Apache Kafka 中添加的分区答案

【问题标题】：Consuming from added partitions in Apache Kafka使用 Apache Kafka 中添加的分区
【发布时间】：2017-09-18 17:12:16
【问题描述】：

我有一个KafkaConsumer 并使用手动分区分配。为了分配分区，我定期使用consumer.partitionsFor(topicId) 来检测添加的分区，因为该作业永远运行并且我想支持这种情况。但是，这总是返回分区的初始列表，除非我重新启动消费者。

有没有办法检测消费者添加的分区？投票或听什么？

【问题讨论】：

我也测试了自动分配，似乎也没有检测到添加的分区。
请显示您的自动分配代码，是的，手动无法检测到此更改，但我相信subscribe 可以处理它。
只是consumer.subscribe(singletonList("my_topic"))。然后我检查consumer.assignment()的输出，添加分区后它没有改变。

【解决方案1】：

KafkaConsumer 有一个configuration property "metadata.max.age.ms"，默认为 5 分钟。这意味着，对consumer.partitionsFor 的每次调用都将返回该时间的缓存副本，然后才获取新的元数据。

将属性设置为0 每次都会获取新的元数据。

【讨论】：

【解决方案2】：

这可能是在您通过assignment() 方法查询时正在重新分配分区。你可以睡 5 秒，然后检查这个来测试它。

获取当前分配给该消费者的分区集。如果订阅是通过使用 assign(Collection) 直接分配分区发生的，那么这将简单地返回已分配的相同分区。 如果使用了主题订阅，那么这将给出当前分配给消费者的主题分区集（如果尚未发生分配，或者分区正在重新分配，则可能没有）强>.

【讨论】：