如何使用Confluent Kafka Consumer仅消费主题最新消息？

我编写了一段C#代码用于读取Kafka主题数据，目标是定期仅获取主题中的最新消息，以便用于实时图表展示。但运行代码时，会读取到24小时前的历史数据。我认为需要在代码中配置偏移量，请问如何在Confluent Kafka Consumer中实现？

public void Read_from_Kafka()
{
  try
  {
    var config = new ConsumerConfig
    {
      BootstrapServers = kafka_URI,
      GroupId = "group",
      AutoOffsetReset = AutoOffsetReset.Earliest,
      SecurityProtocol = SecurityProtocol.Ssl,
      SslCaLocation = "path1",
      SslCertificateLocation = "path2",
      SslKeyLocation = "path3",
      SslKeyPassword = "password",
    };

    CancellationTokenSource source = new CancellationTokenSource();
    CancellationToken cancellationToken = source.Token;

    using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
    {
      consumer.Subscribe(topic_name);
      while (!cancellationToken.IsCancellationRequested)
      {
        var consumeResult = consumer.Consume(cancellationToken);
        Kafka_message_total = consumeResult.Message.Value;

        using (StreamWriter sw = File.AppendText(json_log_file))
        {
          sw.WriteLine("JSON: " + Kafka_message_total + " " + Convert.ToString(DateTime.Now));
        }
        System.Threading.Thread.Sleep(2000);
      }
      consumer.Close();
    }
    using (StreamWriter sw = File.AppendText(error_log))
    {
      sw.WriteLine("Stop Kafka " + " " + Convert.ToString(DateTime.Now));
    }
  }

  catch (Exception ex)
  {
    using (StreamWriter sw = File.AppendText(error_log))
    {
      sw.WriteLine("Kafka Read Error: " + ex + " " + Convert.ToString(DateTime.Now));
    }
  }
}

更新1：我尝试将AutoOffsetReset设置为AutoOffsetReset.Latest，但仍会读取历史数据，该设置无法满足我的需求。

解决方案 #

原因说明 #

AutoOffsetReset 仅在消费者组没有任何已提交的偏移量记录时才会生效。如果你的GroupId之前已经提交过偏移量（哪怕是很早的历史偏移），Kafka会直接从该偏移位置开始消费，完全忽略AutoOffsetReset的配置，这就是你设置Latest后仍读到旧数据的核心原因。

具体实现方案 #

方案1：改用全新的消费者组ID

直接修改GroupId为一个从未使用过的唯一值，这样Kafka会判定这是新的消费者组，此时AutoOffsetReset = AutoOffsetReset.Latest就会生效，启动后直接从主题的最新消息位置开始消费。

修改配置代码：

var config = new ConsumerConfig
{
  BootstrapServers = kafka_URI,
  GroupId = "real-time-chart-group-001", // 替换为新的唯一组ID
  AutoOffsetReset = AutoOffsetReset.Latest,
  // 其余SSL配置保持不变
};

方案2：手动将偏移量重置到分区末尾（保留原GroupId）

如果必须保留原有的GroupId，可以在订阅主题后，手动将每个分区的偏移量设置为当前最新位置（分区末尾）。

修改订阅后的代码逻辑：

consumer.Subscribe(topic_name);

// 获取当前订阅的所有分区
var assignedPartitions = consumer.Assignment;
if (assignedPartitions.Any())
{
  var targetOffsets = new List<TopicPartitionOffset>();
  foreach (var partition in assignedPartitions)
  {
    // 查询分区的最新偏移量（High表示末尾位置）
    var watermarkOffsets = consumer.QueryWatermarkOffsets(partition, TimeSpan.FromSeconds(5));
    targetOffsets.Add(new TopicPartitionOffset(partition, watermarkOffsets.High));
  }
  // 重置消费者偏移量到指定位置
  consumer.Seek(targetOffsets);
}

// 原有消费循环逻辑保持不变
while (!cancellationToken.IsCancellationRequested)
{
  var consumeResult = consumer.Consume(cancellationToken);
  // ... 其余代码
}

方案3：禁用自动提交偏移量（适合纯实时场景）

如果你的场景只需要最新消息，不需要续接之前的消费进度，可以关闭自动提交偏移量，配合新GroupId和AutoOffsetReset.Latest，每次启动都会从最新位置开始。

在配置中添加：

EnableAutoCommit = false,

额外优化建议 #

• 代码中的Thread.Sleep(2000)会强制消费者每2秒才拉取一次数据，严重影响实时性。Consume()方法本身是阻塞式的，会自动等待新消息到来再返回，直接移除Sleep即可实现实时消费。
• 确保Confluent.Kafka NuGet包版本与你的Kafka集群版本兼容，避免出现偏移量相关的兼容性问题。

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