Node.js+Kafka：如何删除消费者已成功提交的数据？

首先得明确你要删除的是哪类数据：

1. 从Kafka Topic中删除已提交的消息 #

Kafka本身不支持基于消费者提交offset来触发消息删除，消息的清理由Topic的配置规则控制，不用在生产者或消费者端加额外功能。你可以通过修改Topic的以下配置实现自动清理：

• retention.ms：设置消息的最大留存时间，到期自动删除
• retention.bytes：设置Topic允许存储的最大数据量，超出后自动清理旧消息

2. 从业务存储（如数据库）中删除对应数据 #

这种情况需要在消费者端添加业务逻辑，在确认消息处理成功后执行删除操作。比如在你的代码里，可以在提交offset前后（确保消息处理完成）加入删除逻辑：

await consumer.run({
    autoCommit: false,
    eachMessage: async ({ topic, partition, message }) => {
        console.log(`RVD Msg ${message.value} on partition ${partition}`);
        // 这里添加业务层面的删除逻辑，比如调用数据库删除接口
        // await deleteBusinessData(message.value);
        // 确认删除成功后提交offset
        await consumer.commitOffsets([{ topic, partition, offset: (Number(message.offset) + 1).toString() }]);
    },
});

总结：如果是清理Kafka集群里的消息，修改Topic配置即可；如果是业务数据删除，在消费者端添加对应业务逻辑。

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