使用Confluent.Kafka对接Azure Event Hubs：可生产消息但无法消费

问题背景 #

生产者配置可正常发送消息到Azure Kafka（Azure门户可见消息），但消费者调用consumer.Consume(cts.Token)时无法获取消息，调试日志显示：

[Confluent.Kafka] Debug: [thrd:main]: Topic anothertest [2]: stored offset INVALID (leader epoch -1), committed offset INVALID (leader epoch -1): not including in commit
[Confluent.Kafka] Debug: [thrd:main]: anothertest [1]: skipping offset validation for offset 7 (leader epoch -1): no leader epoch set
[Confluent.Kafka] Debug: [thrd:main]: anothertest [2]: broker is down: re-query

可正常工作的生产者配置 #

var config = new ProducerConfig
{
    BootstrapServers = bootstrapServers,
    SecurityProtocol = SecurityProtocol.SaslSsl,
    SaslMechanism = SaslMechanism.Plain,
    SaslUsername = saslUsername,
    SaslPassword = saslPassword
};

using (var producer = new ProducerBuilder<Null, string>(config).Build())
{
    Console.WriteLine("Enter a message to send to the topic (or 'exit' to quit):");
    string message;
    while ((message = Console.ReadLine()) != "exit")
    {
        try
        {
            var result = await producer.ProduceAsync(topic, new Message<Null, string> { Value = message });
            Console.WriteLine($"Message '{message}' sent to topic '{topic}' at offset {result.Offset}");
        }
        catch (ProduceException<Null, string> e)
        {
            Console.WriteLine($"Delivery failed: {e.Error.Reason}");
        }
    }
}

无法接收消息的消费者配置片段 #

var config = new ConsumerConfig
{
    BootstrapServers = bootstrapServers,
    SecurityProtocol = SecurityProtocol.SaslSsl,
    SaslMechanism = SaslMechanism.Plain,
    SaslUsername = saslUsername,
    SaslPassword = saslPassword,
    GroupId = consumerGroup,
    SslEndpointIdentificationAlgorithm = SslEndpointIdentificationAlgorithm.None, // Disable endpoint identification
    //SessionTimeoutMs = 6000,  // To reduce disconnection frequency
    //MaxPollIntervalMs = 10000,
    //SocketKeepaliveEnable = true,
    Debug = "all", // Enables detailed logs
    AutoOffsetReset = AutoOffsetReset.Latest,
    EnableAutoCommit = true,
    SocketKeepaliveEnable = true
};

// ...

consumer.Subscribe(topic);

CancellationTokenSource cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
    e.Cancel = true;
    cts.Cancel();
};

try
{
    while (true)
    {
        try
        {
            var consumeResult = consumer.Consume(cts.Token);
            string logMessage = $"Consumed message '{consumeResult.Message.Value}' at: '{consumeResult.TopicPartitionOffset}'.";
            // 缺少日志输出与异常处理代码
        }
    }
}

排查思路与建议 #

1. 补全消费者代码的关键逻辑 #

从提供的代码片段看，存在两处关键缺失：

• 没有输出消费结果的日志（比如Console.WriteLine(logMessage)），即使消费成功也无法看到反馈
• 缺少Consume操作的异常捕获，无法排查潜在错误

补全后的核心消费逻辑示例：

while (true)
{
    try
    {
        var consumeResult = consumer.Consume(cts.Token);
        string logMessage = $"Consumed message '{consumeResult.Message.Value}' at: '{consumeResult.TopicPartitionOffset}'.";
        Console.WriteLine(logMessage); // 必须添加输出才能确认消费结果
    }
    catch (ConsumeException e)
    {
        Console.WriteLine($"Consume error: {e.Error.Reason}");
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Consumer cancelled");
        break;
    }
}

2. 验证偏移量与消费者组配置 #

• 偏移量重置策略：当前设置AutoOffsetReset = AutoOffsetReset.Latest，意味着消费者只会读取启动后新产生的消息。如果测试时先发送消息再启动消费者，自然收不到历史消息。可改为AutoOffsetReset = AutoOffsetReset.Earliest尝试读取所有未消费消息。
• 消费者组ID：如果该消费者组已提交过偏移量且已到最新位置，即使改为Earliest也不会重新读取。可以换一个新的消费者组ID测试，或在Azure门户手动重置该组的偏移量。

3. 解决Broker连接异常 #

日志中broker is down: re-query说明消费者无法正常连接到分区2的Broker，可从以下方向排查：

• 确认BootstrapServers与生产者使用的完全一致（Azure Kafka格式通常为<cluster-name>.azurekafka.chinacloudapi.cn:9093）
• 移除SslEndpointIdentificationAlgorithm = SslEndpointIdentificationAlgorithm.None，Azure Kafka要求严格SSL验证，使用默认的Https值才能通过连接校验
• 检查消费者所在环境的网络权限：是否能访问9093端口，是否有防火墙、代理拦截流量

4. 检查主题分区状态 #

登录Azure门户查看目标Kafka主题的分区状态，确认所有分区的Leader节点正常运行，没有离线或不可用的情况。如果分区2的Leader处于异常状态，消费者将无法从该分区读取消息。

5. 优化消费者配置参数 #

尝试添加或调整以下参数提升连接稳定性：

SessionTimeoutMs = 30000, // Azure Kafka官方建议的会话超时值
MaxPollIntervalMs = 300000,
AutoCommitIntervalMs = 5000,
EnablePartitionEof = true // 检测分区末尾，避免无限阻塞

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