C# Kafka IConsumer的Subscribe方法无法消费消息，但Assign方法正常

我尝试用以下C#代码创建简单Kafka消费者：

private static CancellationTokenSource StartConsumer(IAdminClient client, string topicName)
{
    ConsumerConfig config = new()
    {
        BootstrapServers = BootstrapServers,
        GroupId = "testConsumerGroup",
        AutoOffsetReset = AutoOffsetReset.Earliest,

    };
    IConsumer<Null, string> consumer = new ConsumerBuilder<Null, string>(config).Build();
    //consumer.Assign(new TopicPartitionOffset(topicName, 0, Offset.Beginning));//此代码可以正常工作
    consumer.Subscribe(topicName);//此代码无法工作

    CancellationTokenSource cancellationTokenSource = new();
    CancellationToken cancellationToken = cancellationTokenSource.Token;
    Task.Run(() =>
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            ConsumeResult<Null, string> consumeResult = consumer.Consume();
            Console.WriteLine($"{consumeResult.Offset}: {consumeResult.Message.Value}");
        }
        consumer.Close();
        consumer.Dispose();
    });

    return cancellationTokenSource;
}

当使用Assign方法时，消费者能正常消费消息，但使用Subscribe方法时，完全无法消费，consumer.Consume()一直阻塞不返回。调试发现调用consumer.Subscribe(topicName)后，consumer.Assignment列表是空的，推测是Kafka协调器没有给消费者分配分区。

创建Topic的代码如下：

private static async Task<string> CreateTopic(IAdminClient client, string topicName)
{
    await client.CreateTopicsAsync(new TopicSpecification[]
    {
        new TopicSpecification()
        {
            Name = topicName,
            ReplicationFactor = 1,
            NumPartitions = 1
        }
    });
    return topicName;
}

系统信息：

• 操作系统：Windows 10
• Kafka版本：3.4.0
• Java版本：jdk 1.8.0_202（32位）
• Confluent.Kafka NuGet包版本：2.0.2

解决方法 #

1. 等待Topic元数据同步 #

调用CreateTopicsAsync后，Kafka需要时间完成Topic创建和元数据同步，直接启动消费者可能导致无法获取Topic信息。在创建Topic后验证Topic存在再启动消费者：

private static async Task<string> CreateTopic(IAdminClient client, string topicName)
{
    await client.CreateTopicsAsync(new TopicSpecification[]
    {
        new TopicSpecification()
        {
            Name = topicName,
            ReplicationFactor = 1,
            NumPartitions = 1
        }
    });
    
    // 循环验证Topic是否创建完成
    bool topicExists = false;
    int retryCount = 0;
    while (!topicExists && retryCount < 5)
    {
        var metadata = client.GetMetadata(TimeSpan.FromSeconds(2));
        topicExists = metadata.Topics.Any(t => t.Topic == topicName);
        if (!topicExists)
        {
            await Task.Delay(1000);
            retryCount++;
        }
    }
    
    return topicName;
}

2. 完善消费者配置 #

确保BootstrapServers配置正确，显式配置必要参数并开启调试日志排查问题：

ConsumerConfig config = new()
{
    BootstrapServers = BootstrapServers,
    GroupId = "testConsumerGroup",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = true,
    LogLevel = LogLevel.Debug // 开启调试日志，查看分区分配细节
};

3. 更换64位JDK #

32位JDK可能限制Kafka稳定性，更换为64位JDK（推荐JDK 11或兼容的64位JDK 8），重启Kafka集群后测试。

4. 手动刷新元数据 #

在Subscribe后强制刷新元数据，确保消费者获取Topic分区信息：

consumer.Subscribe(topicName);
// 强制刷新元数据，等待5秒超时
var metadata = consumer.GetMetadata(topicName, TimeSpan.FromSeconds(5));

5. 重置消费者组偏移量 #

如果消费者组存在异常偏移量，通过Kafka命令行工具重置：

kafka-consumer-groups.bat --bootstrap-server <你的Kafka地址> --group testConsumerGroup --reset-offsets --to-earliest --topic <目标Topic名称> --execute

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