消费者提交Offset后仍接收重复消息问题排查求助

问题描述 #

消费者服务已提交Offset，但仍接收到重复消息，Pod重启或CI/CD部署场景下该问题尤为突出，已通过每2分钟重启服务的方式复现场景。

已尝试方案 #

• 初始采用间隔式自动提交（配置autoCommitInterval和autoCommitThreshold），问题未解决
• 改为Offset级手动提交，重复消息问题依然存在

当前配置 #

消费者初始化 #

const groupId = `group-${topicName.toLowerCase()}-${process.env.NODE_ENV}`;

多个服务实例共用同一groupId

订阅配置 #

await this.consumer.subscribe({
     topic,
     fromBeginning: true,
});

消息消费逻辑 #

this.consumer.run({
    eachBatchAutoResolve: false,
    // autoCommitInterval: 1000,
    // autoCommitThreshold: 5,
    eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning, isStale }) => {
                if (!isRunning() || isStale()) return;
                await heartbeat();
                await processTopicMsg(logger, batch.messages, country, resolveOffset, heartbeat);
                await heartbeat();
            },
        });

消息处理函数（processTopicMsg） #

resolveOffset(topicMsg[Index].offset);
await heartbeat();

问题根源 #

1. fromBeginning: true的致命问题：每次服务启动都会强制从topic起始位置消费，直接覆盖Kafka集群中该groupId保存的已提交Offset，导致重启后重复消费历史消息。
2. 手动提交不完整：当前仅调用resolveOffset标记消息为已处理，但未显式触发Offset提交到集群，部分Kafka客户端（如kafkajs）中resolveOffset仅为本地标记，不会同步到集群。
3. Offset提交逻辑错误：提交的Offset应为下一条待消费消息的偏移量，而非当前处理消息的偏移量，否则重启后可能重复消费已处理的最后一条消息。
4. 优雅关闭缺失：Pod重启时未等待消费者完成Offset提交就终止进程，导致集群未更新最新Offset。

修复方案 #

1. 修正订阅配置 #

将fromBeginning改为false，让消费者依赖Kafka集群保存的groupId Offset启动，仅在groupId首次创建时才从起始位置消费：

await this.consumer.subscribe({
     topic,
     fromBeginning: false,
});

若需首次启动初始化消费位置，可通过脚本预先在集群中设置该groupId的初始Offset，或通过代码判断groupId是否存在来动态设置fromBeginning。

2. 完善手动提交逻辑 #

在处理完所有消息后，显式提交正确的Offset到集群，同时确保单条消息处理成功后再标记：

this.consumer.run({
    eachBatchAutoResolve: false,
    eachBatch: async ({ batch, resolveOffset, commitOffsets, heartbeat, isRunning, isStale }) => {
        if (!isRunning() || isStale()) return;
        
        try {
            // 逐条处理消息，确保每条处理成功
            for (const msg of batch.messages) {
                await heartbeat();
                // 处理单条消息，确保处理逻辑无异常
                await processSingleMsg(logger, msg, country);
                resolveOffset(msg.offset);
            }
            
            // 提交下一条待消费的Offset（当前批次最后一条消息偏移量+1）
            await commitOffsets([{
                topic: batch.topic,
                partition: batch.partition,
                offset: (parseInt(batch.messages[batch.messages.length - 1].offset) + 1).toString()
            }]);
            await heartbeat();
        } catch (err) {
            // 处理失败时可根据业务选择重试或跳过，避免阻塞消费
            logger.error('消息批次处理失败', err);
        }
    },
});

3. 实现优雅关闭 #

注册进程退出钩子，确保消费者在Pod重启前完成Offset提交并断开连接：

process.on('SIGTERM', async () => {
    logger.info('收到终止信号，开始优雅关闭消费者');
    await this.consumer.disconnect();
    process.exit(0);
});

process.on('SIGINT', async () => {
    logger.info('收到中断信号，开始优雅关闭消费者');
    await this.consumer.disconnect();
    process.exit(0);
});

4. 验证多实例消费一致性 #

确保多个共用同一groupId的实例消费逻辑一致，避免因部分实例提交Offset滞后导致的集群Offset回退。

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