Azure 事件中心：Kafka 消费者在 Java 中不进行任何轮询

Question

我正在尝试从 azure event-hub 读取消费者中的消息

这是我的消费者配置：

private final KafkaConsumer<Integer, String> consumer;
private final String topic;

public SampleConsumer(String topic) throws IOException {
Properties props = new Properties();
props.load(new FileReader("src/main/resources/consumer.config"));
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
    "MyNamespace.servicebus.windows.net:9093");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "GROUP_ID");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.IntegerDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.StringDeserializer");

consumer = new KafkaConsumer<>(props);
this.topic = topic;
}

这是我的 consumer.config 文件：

bootstrap.servers=<MyNamespace>.servicebus.windows.net:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required 
username="$**********" password="*******";

最后在这个方法中我尝试读取记录：

@Override
public void doWork() {
  consumer.subscribe(Collections.singletonList(this.topic));
  ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(1000));
  System.out.println(records.count());
  for (ConsumerRecord<Integer, String> record : records) {
    System.out.println("Received message: (" + record.value());
  }
}

它正确连接到 azure 和 event-hub。 然后它尝试读取记录，但每次它返回 0 而没有任何错误消息。我在互联网上搜索并使用了很多不同的解决方案，但到目前为止没有任何效果。 event-hub 中至少应该有一些记录。

有人可以帮我找出问题所在吗？

Answer 1

您是否正在积极制作唱片？如果没有，那么也许您应该从主题的开头开始消费，将auto.offset.reset=earliest 设置为新的组 id

默认行为是在主题结束时开始并在消费者开始之后轮询新事件