使用Confluent Kafka的.NET C#项目：如何从三分区Topic读取最新消息？

我在.NET C#项目中使用Confluent Kafka库，当前代码只能读取指定分区的最新消息，但我的Topic配置了0、1、2三个分区，单个分区的最新消息不一定是数据源发送到Kafka的全局最新消息。请问怎么调整代码适配多分区？Confluent Kafka有没有内置简便方法？还是必须从所有分区的Offset.End位置读取消息，再通过时间戳判断哪条是全局最新？

原代码：

CancellationTokenSource source = new CancellationTokenSource();
CancellationToken cancellationToken = source.Token;
using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
{
    consumer.Subscribe("My_Topic");               

    while (true)
    {
        TopicPartitionOffset tps = new TopicPartitionOffset(new TopicPartition("My_Topic", 1), Offset.End);
        consumer.Assign(tps);
        var consumeResult = consumer.Consume(cancellationToken);                      
        
        Kafka_message_total = consumeResult.Message.Value;

        // additional code to send the message value to an application

        System.Threading.Thread.Sleep(2000);

    }

    consumer.Close();
}

Confluent Kafka没有直接获取全局最新消息的内置函数，因为Kafka的分区是独立存储的，全局最新需要自行判断。可以通过以下步骤实现：

• 获取Topic的所有分区
  通过consumer.GetPartitionsForTopic方法自动获取目标Topic的所有分区，无需手动指定分区编号。

• 消费每个分区的最新消息
  遍历所有分区，为每个分区设置Offset.End（最新位置），消费该位置的消息并记录每条消息的时间戳和内容。

• 筛选全局最新消息
  比较所有分区最新消息的时间戳，取时间戳最大的那条即为全局最新消息。

调整后的代码示例：

CancellationTokenSource source = new CancellationTokenSource();
CancellationToken cancellationToken = source.Token;
using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
{
    string topicName = "My_Topic";
    // 获取当前Topic的所有分区
    var partitions = consumer.GetPartitionsForTopic(topicName);

    while (!cancellationToken.IsCancellationRequested)
    {
        List<ConsumeResult<Ignore, string>> latestPartitionMessages = new List<ConsumeResult<Ignore, string>>();

        foreach (var partition in partitions)
        {
            // 定位到当前分区的最新偏移量位置
            var targetOffset = new TopicPartitionOffset(partition, Offset.End);
            consumer.Assign(targetOffset);
            
            try
            {
                // 消费该分区的最新消息
                var consumeResult = consumer.Consume(cancellationToken);
                if (consumeResult != null && consumeResult.Message != null)
                {
                    latestPartitionMessages.Add(consumeResult);
                }
            }
            catch (ConsumeException ex)
            {
                // 处理消费异常，比如分区不可用等情况
                Console.WriteLine($"消费分区{partition.Partition}失败: {ex.Message}");
            }
        }

        if (latestPartitionMessages.Any())
        {
            // 按时间戳倒序排序，取第一条作为全局最新消息
            var globalLatest = latestPartitionMessages.OrderByDescending(m => m.Message.Timestamp.UtcDateTime).First();
            Kafka_message_total = globalLatest.Message.Value;

            // 后续业务逻辑：将消息发送到应用
            // ...
        }

        System.Threading.Thread.Sleep(2000);
    }

    consumer.Close();
}

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