Kafka消费者组未提交消息丢失问题及不丢消息保障诉求

看起来你已经明确了核心需求：基于单wurstmeister Kafka Broker + 单消费者组单消费者的架构，通过手动控制Offset提交来确保消息零丢失——当消息处理失败（比如内容为'error'）时不提交Offset，让消费者持续重试该消息，直到处理成功。下面是具体实现方案和测试步骤：

一、消费者核心配置（必选） #

要实现手动Offset提交，首先必须关闭自动提交，并配置必要的参数以避免消息丢失：

# 关闭自动提交（核心）
enable.auto.commit=false
# 当消费者组无初始Offset时，从最早的消息开始消费（避免遗漏历史消息）
auto.offset.reset=earliest
# 固定消费者组ID
group.id=my-single-consumer-group
# 其他基础配置（示例）
bootstrap.servers=localhost:9092
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

二、代码实现（以Java Kafka客户端为例） #

核心逻辑是：拉取消息后，逐个处理，仅当处理成功时才手动提交Offset；遇到'error'消息时，跳过提交，让消费者在下一次poll时重新拉取该消息。

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class ManualOffsetConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        // 加载核心配置
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-single-consumer-group");
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (var record : records) {
                    String message = record.value();
                    System.out.printf("Received message: %s, offset: %d%n", message, record.offset());

                    try {
                        // 模拟消息处理逻辑
                        processMessage(message);
                        // 处理成功，手动提交当前消息的Offset（同步提交，保证提交成功）
                        consumer.commitSync(Collections.singletonMap(
                                new TopicPartition(record.topic(), record.partition()),
                                new OffsetAndMetadata(record.offset() + 1)
                        ));
                        System.out.printf("Committed offset for message: %s%n", message);
                    } catch (Exception e) {
                        // 处理失败（比如消息是'error'），不提交Offset，下次会重新消费
                        System.err.printf("Failed to process message: %s, will retry. Error: %s%n", message, e.getMessage());
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }

    private static void processMessage(String message) throws Exception {
        if ("error".equals(message)) {
            // 模拟处理失败，抛出异常
            throw new Exception("Message content is 'error', processing failed");
        }
        // 正常消息处理逻辑
        System.out.printf("Successfully processed message: %s%n", message);
    }
}

关键逻辑说明： #

• 手动提交Offset：使用commitSync同步提交，确保Offset成功写入Kafka后再继续，避免提交丢失。
• 失败不提交：当消息是'error'时，抛出异常进入catch块，不执行commitSync，消费者下次poll会从上次未提交的Offset位置重新拉取这条消息，实现永久重试。
• 精准提交：针对单个消息的Offset提交（record.offset() + 1），确保只有当前消息处理成功才推进Offset，不会因为批量提交导致部分失败消息被跳过。

三、测试步骤（使用kafka-console-producer） #

1. 启动wurstmeister Kafka Broker：确保Broker正常运行，监听端口9092。
2. 创建测试Topic（如果未创建）：

docker exec -it <kafka-container-id> kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

3. 启动消费者程序：运行上面的Java代码，控制台会输出消费日志。
4. 发送测试消息：
   • 先发送正常消息：

     docker exec -it <kafka-container-id> kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic
     >hello
     >world
     

     观察消费者控制台，会看到消息被处理并提交Offset，不会重复消费。
   • 发送'error'消息：

     >error
     

     此时消费者会打印失败日志，并且会持续重复拉取这条'error'消息，直到你修改处理逻辑让它成功为止——完全符合你要求的"永久重试、不丢消息"的规则。

四、注意事项 #

• 避免消费者停机：如果消费者因故障停机，重启后会从上次未提交的Offset位置继续消费，不会丢失消息。
• 单消费者场景：因为你使用单消费者组单消费者，不存在Rebalance问题，Offset提交逻辑会更稳定。
• wurstmeister Broker配置：确保Broker的ADVERTISED_LISTENERS配置正确，比如设置为PLAINTEXT://localhost:9092，避免消费者无法连接。

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