使用Kafka API无法消费Azure Event Hub消息的问题求助

尝试使用kafka-python的KafkaConsumer读取Azure Event Hub消息，但无法建立连接，bootstrap_connected()始终返回False，代码卡在消息循环前。使用Azure官方的EventHubConsumerClient可以正常接收消息，但业务要求必须使用Kafka Consumer Client。

提供的代码：

from kafka import KafkaConsumer
from json import loads
consumer = KafkaConsumer(<EVENTHUB_NAME>,
                         bootstrap_servers='<NAME>.servicebus.windows.net:9093',
                         auto_offset_reset='earliest',
                         enable_auto_commit=True,
                         security_protocol='SASL_SSL',
                         sasl_mechanism='PLAIN',
                         sasl_plain_username=<CLIENT ID>,
                         sasl_plain_password=<CONNECTION_STRING>,
                         group_id=<CONSUMER_GROUP_NAME>,
                         api_version=(0, 11),
                         value_deserializer=lambda x: loads(x.decode('utf-8')))
print(consumer) #returns <kafka.consumer.group.KafkaConsumer object at 0x00000000012A8130>
x = consumer.bootstrap_connected() #returns False
print(x)
for message in consumer:
    message = message.value
    print('{} added'.format(message))

以下是针对连接失败的排查和修复步骤：

• 修正SASL认证参数
  Azure Event Hub的Kafka兼容模式下，SASL认证的username固定为$ConnectionString，而非客户端ID；password才是完整的Event Hub连接字符串（格式类似Endpoint=sb://<NAME>.servicebus.windows.net/;SharedAccessKeyName=<KEY_NAME>;SharedAccessKey=<KEY_VALUE>）。这是连接失败的最常见原因，修改后的SASL配置应为：

  sasl_plain_username='$ConnectionString',
  sasl_plain_password='<YOUR_FULL_EVENTHUB_CONNECTION_STRING>',
  

• 移除过时的API版本指定
  api_version=(0,11)对应非常旧的Kafka版本，Azure Event Hub最低支持Kafka 1.0.0版本，指定过时版本会引发兼容性问题，建议删除这个参数，让客户端自动协商适配版本。

• 检查消费者组有效性
  确保group_id对应的消费者组在Event Hub中存在，默认消费者组为$Default，如果使用自定义组，需确认已在Azure Portal中创建。

• 验证网络连通性
  检查运行代码的环境是否能访问*.servicebus.windows.net:9093端口，部分防火墙或代理可能会阻止该端口的SSL连接，可通过telnet <NAME>.servicebus.windows.net 9093或nc -zv <NAME>.servicebus.windows.net 9093测试连通性。

• 启用调试日志排查细节
  添加日志配置，查看连接过程中的具体错误信息：

  import logging
  logging.basicConfig(level=logging.DEBUG)
  

  调试日志会显示SASL认证过程、Broker连接尝试等细节，帮助定位具体失败原因。

• 修正后的完整示例代码

  from kafka import KafkaConsumer
  from json import loads
  import logging
  
  # 启用日志（可选，用于排查问题）
  logging.basicConfig(level=logging.INFO)
  
  consumer = KafkaConsumer(
      '<EVENTHUB_NAME>',
      bootstrap_servers='<NAME>.servicebus.windows.net:9093',
      auto_offset_reset='earliest',
      enable_auto_commit=True,
      security_protocol='SASL_SSL',
      sasl_mechanism='PLAIN',
      sasl_plain_username='$ConnectionString',
      sasl_plain_password='Endpoint=sb://<NAME>.servicebus.windows.net/;SharedAccessKeyName=<KEY_NAME>;SharedAccessKey=<KEY_VALUE>',
      group_id='<CONSUMER_GROUP_NAME>',
      value_deserializer=lambda x: loads(x.decode('utf-8'))
  )
  
  print("Consumer initialized, waiting for messages...")
  for message in consumer:
      print(f"{message.value} added")
  

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