时间戳

首页 > 解决方案 > Kafka 消费者分配返回空集

java - Kafka 消费者分配返回空集

问题描述

我已经订阅了一个 Kafka 主题，如下所示。只有在为消费者分配了分区后，我才需要运行一些逻辑。

然而consumer.assignment()，无论我等多久，它都会以空集的形式返回。如果我没有 while 循环然后做一个 consumer.poll()我确实从主题中获取记录。有人能告诉我为什么会这样吗？

    consumer.subscribe(topics);
    Set<TopicPartition> assigned=Collections.emptySet();
    while(isAssigned) {
          assigned = consumer.assignment();
          if(!assigned.isEmpty()) {
              isAssigned= false;
          }
    }

//consumer props
Properties props = new Properties();
        props.put("bootstrap.servers", "xxx:9092,yyy:9092");
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "false");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
        props.put("schema.registry.url", "http://xxx:8081");
        props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put("max.poll.records", "100");

标签： javaapache-kafkakafka-consumer-api

解决方案

在你打电话之前poll()，消费者只是在闲置。

只有在poll()被调用后，它才会启动与集群的连接，获取分配的分区并尝试获取消息。

Javadoc中提到了这一点：

订阅一组主题后，消费者会在调用 poll(Duration) 时自动加入群组。

一旦您开始调用poll()，您可以使用assignment()甚至注册 aConsumerRebalanceListener以在从您的消费者实例分配或撤销分区时收到通知。

推荐阅读