首页 文章

Kafka 模式订阅 . 新主题没有触发重新 balancer

提问于
浏览 次
5

根据kafka javadocs的文件，如果我：

  • 订阅模式

  • 创建与模式匹配的主题

应该发生重新 balancer ，这使消费者从该新主题中读取 . 但那并没有发生 .

如果我停止并启动消费者，它确实会选择新主题 . 所以我知道新主题与模式匹配 . 在_2793814中可能有这个问题的重复，但这个问题无处可去 .

我看到kafka日志并没有错误，它只是不会触发重新 balancer . 当消费者加入或死亡时触发重新 balancer ，但是在创建新主题时不会触发（即使分区被添加到现有主题，但这是另一个主题） .

我正在使用kafka 0.10.0.0和“新消费者API”的官方Java客户端，意思是代理GroupCoordinator而不是胖客户端zookeeper .

这是示例消费者的代码：

public class SampleConsumer {
public static void main(String[] args) throws IOException {
    KafkaConsumer<String, String> consumer;
    try (InputStream props = Resources.getResource("consumer.props").openStream()) {
        Properties properties = new Properties();
        properties.load(props);
        properties.setProperty("group.id", "my-group");

        System.out.println(properties.get("group.id"));
        consumer = new KafkaConsumer<>(properties);
    }
    Pattern pattern = Pattern.compile("mytopic.+");
    consumer.subscribe(pattern, new SampleRebalanceListener());
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("%s %s\n", record.topic(), record.value());
        }
    }
}

}

在制作人中，我正在向名为mytopic1，mytopic2等的主题发送消息 .

如果不触发重新 balancer ，模式几乎没用 .

你知道为什么没有发生重新 balancer 吗？

java apache-kafka kafka-consumer-api

1 回答

  • 6

    文档提到“模式匹配将定期针对检查时存在的主题进行 . ”事实证明，“周期性”对应于metadata.max.age.ms属性 . 通过将该属性（在我的代码示例中的“consumer.props”内）设置为5000，我可以看到它每5秒检测一次新主题和分区 .

    这是根据设计，根据这个jira票https://issues.apache.org/jira/browse/KAFKA-3854：

    关于JIRA的最后一点，指出后来创建的与消费者订阅模式匹配的主题在创建时不会分配给消费者，这似乎与设计的一致 . 处理该情况需要重复订阅相同模式的subscribe（） .

    刷新元数据轮询执行故障单中提到的“重复订阅（）” .

    这是令人困惑的来自Kafka 0.8，其中真正的触发基于zookeper Watch ，而不是轮询 . 对于这种情况，IMO 0.9更多地是降级，而不是“及时”重新 balancer ，这变成了高频率轮询与开销，或者在对新主题/分区作出反应之前长时间进行低频率轮询 .

    回复于

相关问题