java - 为什么调用Kafka的seekToBeginning和seekToEnd API永远挂起？

Question

我的环境:

• Kafka 版本:kafka_2.10-0.10.0.0
• Kafka Java API 版本(客户端):kafka-clients-0.10.0.0

我的配置:

• 主题:event_notification
• 分区:20
• 客户端消费者线程:1
• 消费者组 ID:event_cg01
• 自动提交标志:false

根据要求，在我的应用程序启动期间，我必须根据标志将偏移量设置为开始或结束。为此，我使用以下代码:

final List<PartitionInfo> partitionsInfos = kafkaConsumer.partitionsFor(this.topic);

final List<TopicPartition> assignedPartitions = FluentIterable
      .from(partitionsInfos)                                                             
      .filter(Predicates.notNull())
      .transform(new Function<PartitionInfo, TopicPartition>() {                                                           
          @Override
          public TopicPartition apply(final PartitionInfo input) {                                                                         
            return new TopicPartition(topic, input.partition());
          }
      }).toList();

switch (listenMode) {
case OLDEST:
  kafkaConsumer.seekToBeginning(assignedPartitions);
  break;
case LATEST:
  kafkaConsumer.seekToEnd(assignedPartitions);
  break;
default:
  break;
}

此代码未按预期工作。它永远卡在 seekToBeginning 和 seekToEnd 调用上。

我错过了什么吗？

Answer 1

在使用 seek() 之前，您首先需要 subscribe() 某个主题或 assign() 主题的分区消费者。请注意，subscribe() 和 assign() 是惰性调用，因此，您还需要对 进行“虚拟调用” >poll()，然后才能使用seek() 或seekToBeginning() 或seekToEnd()。