如何让Kafka重试消费者仅处理初始化时已存在的消息

问题 #

我维护两个Kafka Topic：

• 第一个Topic由实时消费者持续处理消息，处理失败的消息会被发布到第二个重试Topic
• 重试Topic由独立消费者每日消费一次，处理失败的消息会重新发布回该重试Topic

现在遇到的问题是：我希望重试消费者只处理启动初始化时重试Topic中已存在的消息，忽略后续新发布的消息，避免消息因处理失败陷入「重复发布-处理」的无限循环。

我尝试了以下代码，但重试消费者仍然会消费新消息：

_kafkaConsumer= consumerBuilder.Build();

_kafkaConsumer.Subscribe(TopicName);

// Wait for the consumer to be assigned partitions
while (_kafkaConsumer.Assignment.Count == 0)
{
    _kafkaConsumer.Consume(TimeSpan.FromMilliseconds(100));
}

// Store the high watermark offsets for each partition at the time of subscription
_endOffsets = new Dictionary<TopicPartition, long>();
foreach (var partition in _kafkaConsumer.Assignment)
{
    var watermarkOffsets = _kafkaConsumer.QueryWatermarkOffsets(partition, TimeSpan.FromSeconds(10));
    _endOffsets[partition] = watermarkOffsets.High; // Store the high watermark
}

var result = _kafkaConsumer.Consume(new TimeSpan(0, 0, 1));
if (result != null)
{
    return result.Offset < _endOffsets[result.TopicPartition] ? result : null!;
}

请问有没有办法配置Kafka消费者仅处理初始化时已存在的消息、忽略后续新消息？或者有其他方案可以避免消费者重新处理刚发布的消息，留待次日处理？

解决方案 #

1. 修正高水位判断逻辑，消费到指定偏移量后停止分区消费 #

你的代码已经获取了初始化时的高水位，但仅在单次消费时判断，未在循环中持续校验，也没有在达到高水位后停止对应分区的消费。调整后的逻辑如下：

_kafkaConsumer= consumerBuilder.Build();
_kafkaConsumer.Subscribe(TopicName);

// 等待分区分配完成
while (_kafkaConsumer.Assignment.Count == 0)
{
    _kafkaConsumer.Consume(TimeSpan.FromMilliseconds(100));
}

// 记录初始化时每个分区的高水位
var endOffsets = new Dictionary<TopicPartition, long>();
foreach (var partition in _kafkaConsumer.Assignment)
{
    var watermarkOffsets = _kafkaConsumer.QueryWatermarkOffsets(partition, TimeSpan.FromSeconds(10));
    endOffsets[partition] = watermarkOffsets.High;
}

// 循环消费，直到所有分区都达到初始化时的高水位
var completedPartitions = new HashSet<TopicPartition>();
while (completedPartitions.Count < endOffsets.Count)
{
    var result = _kafkaConsumer.Consume(TimeSpan.FromSeconds(5));
    if (result == null) continue;

    var tp = result.TopicPartition;
    // 判断当前消息偏移量是否小于初始化时的高水位
    if (result.Offset < endOffsets[tp])
    {
        // 处理消息逻辑
        ProcessMessage(result);
        // 提交偏移量
        _kafkaConsumer.Commit(result);
    }
    else
    {
        // 该分区已消费到初始化时的末尾，标记为完成并取消订阅
        completedPartitions.Add(tp);
        _kafkaConsumer.Unsubscribe(new List<TopicPartition> { tp });
    }
}

// 所有分区处理完成后关闭消费者
_kafkaConsumer.Close();

2. 基于时间戳过滤消费 #

如果重试Topic的消息按时间顺序写入，可以在消费者启动时记录当前时间，只消费启动时间之前的消息：

var startTime = DateTime.UtcNow;
_kafkaConsumer= consumerBuilder.Build();
_kafkaConsumer.Subscribe(TopicName);

while (true)
{
    var result = _kafkaConsumer.Consume(TimeSpan.FromSeconds(5));
    if (result == null) break;

    // 转换Kafka消息时间戳为UTC时间
    var msgTime = result.Message.Timestamp.UtcDateTime;
    if (msgTime < startTime)
    {
        // 处理消息
        ProcessMessage(result);
        _kafkaConsumer.Commit(result);
    }
    else
    {
        // 跳过启动后新产生的消息
        continue;
    }
}

3. 调整重试消息的发送策略（从根源避免循环） #

给重试失败的消息添加延迟发送属性，直接将失败消息延迟到次日再投递到重试Topic，这样当日的重试消费者不会看到这些新的重试消息：

• 使用ProducerRecord设置延迟时间（需Kafka版本支持，且Topic配置message.timestamp.type=CreateTime）
• 或基于Kafka实现时间轮延迟队列，将失败消息暂存，次日再投递到重试Topic

4. 使用手动分配分区而非自动订阅 #

自动订阅可能在消费过程中触发重新分区分配，导致新分区加入。改为手动分配所有分区，直接获取每个分区的高水位，消费到该偏移量后停止：

// 获取重试Topic的所有分区
var partitions = _kafkaConsumer.PartitionsFor(TopicName)
                               .Select(p => new TopicPartition(TopicName, p.Partition))
                               .ToList();

_kafkaConsumer.Assign(partitions);

// 记录每个分区的高水位并定位到起始位置消费
var endOffsets = new Dictionary<TopicPartition, long>();
foreach (var partition in partitions)
{
    var watermarkOffsets = _kafkaConsumer.QueryWatermarkOffsets(partition, TimeSpan.FromSeconds(10));
    endOffsets[partition] = watermarkOffsets.High;
    _kafkaConsumer.Seek(partition, Offset.Beginning);
}

// 循环消费直到所有分区完成
var completedPartitions = new HashSet<TopicPartition>();
while (completedPartitions.Count < endOffsets.Count)
{
    var result = _kafkaConsumer.Consume(TimeSpan.FromSeconds(5));
    if (result == null) continue;

    var tp = result.TopicPartition;
    if (result.Offset < endOffsets[tp])
    {
        ProcessMessage(result);
        _kafkaConsumer.Commit(result);
    }
    else
    {
        completedPartitions.Add(tp);
        _kafkaConsumer.Unassign(new List<TopicPartition> { tp });
    }
}

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