Java环境Kafka Consumer连接Azure Event Hubs拉取不到消息问题求助

1 配置冲突问题 #

• 你代码中先加载了consumer.config配置文件，后续又手动覆盖了ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG参数，需要确认你硬编码的MyNamespace.servicebus.windows.net:9093已经替换为真实的Event Hub命名空间，否则会出现配置错误
• consumer.config中的sasl.jaas.config配置存在换行错误：Java Properties文件中长属性值换行需要在每行末尾加反斜杠\，你当前的配置会导致username、password属性没有被正确识别为sasl.jaas.config的一部分，正确配置如下：

bootstrap.servers=<真实命名空间>.servicebus.windows.net:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="$ConnectionString" password="<你的Event Hub完整连接字符串>";

注意：Azure Event Hub对接Kafka协议时，sasl的username固定为$ConnectionString，password需要填写完整的Event Hub连接字符串，不要自行替换为其他值

2 消费逻辑错误 #

• 你每次调用doWork方法都会重新执行consumer.subscribe，且只调用一次poll方法：Kafka消费者第一次调用poll时需要完成消费组协调、分区分配、元数据同步等初始化操作，单次1000ms的超时大概率无法完成初始化流程，自然返回0条记录。正确的消费逻辑应该是一次订阅后循环调用poll，示例修改如下：

// 订阅逻辑放到构造方法中，不要每次poll前重复调用
public SampleConsumer(String topic) throws IOException {
    // 原有配置加载逻辑不变
    consumer = new KafkaConsumer<>(props);
    this.topic = topic;
    consumer.subscribe(Collections.singletonList(this.topic));
}

@Override
public void doWork() {
    // 循环拉取，初始化阶段如果拉取为空会自动重试
    while (true) {
        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());
        }
    }
}

3 消费组offset问题 #

• 你配置的AUTO_OFFSET_RESET_CONFIG=earliest仅在当前消费组没有已提交的offset时才生效，如果你之前已经用GROUP_ID这个消费组ID消费过，哪怕之前没有拉取到消息，offset也可能已经被自动提交到了最新位置，存量消息自然不会被重复消费。可以换一个全新的、从未使用过的消费组ID测试验证。

4 基础校验项 #

• 确认代码中传入的topic名称和你Event Hub的实例名称完全一致
• 确认Event Hub的消息保留周期还未过期，存量消息仍然在保留时间范围内

内容的提问来源于stack exchange，提问作者Spaceman