如何重试Kafka未提交消息的处理流程并捕获重试信息？

一、实现消息重试直至提交的核心方案 #

auto.offset.reset是消费者找不到有效偏移量时的兜底重置策略，和当前单条消息重试需求无关。必须采用手动偏移量提交+本地重试循环的方案：

1. 消费者核心配置

   • 强制关闭自动提交：enable.auto.commit=false，完全掌控偏移量提交时机
   • 兜底配置auto.offset.reset=earliest（仅在首次消费或偏移量丢失时生效）
   • 可选设置max.poll.records=1，确保每次只拉取一条消息，避免重试时阻塞其他消息处理

2. 消费逻辑实现

   • 对每条消息单独维护重试计数器
   • 处理失败（不满足条件）时，不提交偏移量，重复执行处理逻辑；处理成功后，手动提交当前消息的偏移量
   • 多线程消费场景下，需保证单条消息的重试逻辑在同一线程内执行，避免偏移量混乱

示例代码（Java）：

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-server:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

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

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        int retryCount = 0;
        boolean processSuccess = false;
        
        // 最多重试5次，之后可选择继续重试或转死信队列
        while (retryCount <= 5) {
            try {
                // 自定义业务校验逻辑
                if (isMessageValid(record.value())) {
                    processSuccess = true;
                    break;
                } else {
                    retryCount++;
                    // 记录重试信息
                    logRetryDetails(record, retryCount, "消息不满足业务条件");
                    // 可选：添加重试间隔，避免高频重试
                    Thread.sleep(1000);
                }
            } catch (Exception e) {
                retryCount++;
                logRetryDetails(record, retryCount, "处理异常：" + e.getMessage());
                Thread.sleep(1000);
            }
        }

        if (processSuccess) {
            // 手动提交当前消息的偏移量（偏移量+1表示已处理完成）
            consumer.commitSync(Collections.singletonMap(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            ));
        } else {
            // 5次重试失败后的处理：可选择继续重试/转死信队列/放弃
            // 不提交偏移量的话，下次poll会重新拉取这条消息
            log.error("消息{}经5次重试仍未通过校验，将继续重试", record.value());
        }
    }
}

二、捕获5次重试相关信息的实现 #

在每次重试时，通过日志系统持久化关键信息，核心记录维度包括：

• 消息元数据：topic、partition、offset、key、value
• 重试次数
• 重试时间戳
• 失败原因（业务不满足/异常堆栈）
• 消费者组ID

示例日志记录方法：

private void logRetryDetails(ConsumerRecord<String, String> record, int retryCount, String reason) {
    Logger logger = LoggerFactory.getLogger(KafkaConsumerDemo.class);
    logger.warn("重试日志 | 消费组: {} | Topic: {} | Partition: {} | Offset: {} | 消息内容: {} | 重试次数: {} | 失败原因: {} | 时间: {}",
        "test-group",
        record.topic(),
        record.partition(),
        record.offset(),
        record.value(),
        retryCount,
        reason,
        LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
    );
}

注意事项 #

• 避免无限重试阻塞消费：5次重试失败后，建议将消息转发到死信队列（DLQ），避免影响正常消息的消费进度
• 单条消息重试时，max.poll.records=1是关键配置，确保重试逻辑不会阻塞其他消息
• 手动提交偏移量时，必须提交record.offset() + 1，否则会重复消费当前消息

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