Java无Spring环境下Kafka如何配置重复接收未提交消息不推进offset

完全可以实现该需求，核心逻辑是通过手动控制消费位点，处理失败时将位点重置回当前消息的offset即可，无需额外修改Kafka服务端配置。

实现前提 #

• 已将enable.auto.commit参数设置为false，关闭自动提交offset能力
• 所有offset提交动作都由业务逻辑手动触发

核心实现逻辑 #

1. 每次拉取到消息后先执行业务处理
2. 处理成功则手动提交当前offset+1的位点，后续拉取会自动读取下一条消息
3. 处理失败/需要重复消费当前消息时，调用consumer.seek()方法将消费位点重置为当前消息的offset，同时终止当前批次消息的遍历，下次poll()时就会重新拉取这条消息，不会向后读取后续offset的消息

代码示例 #

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaRepeatConsumeDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "Kafka服务地址:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "你的消费组ID");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 关闭自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("你的Topic名称"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    // 执行业务处理逻辑，返回处理是否成功
                    boolean processResult = doBusinessProcess(record);
                    if (!processResult) {
                        // 处理失败，重置位点到当前消息的offset
                        consumer.seek(new TopicPartition(record.topic(), record.partition()), record.offset());
                        // 终止当前批次遍历，避免同批次后续消息被处理
                        break;
                    } else {
                        // 处理成功，手动提交offset
                        consumer.commitSync(Collections.singletonMap(
                                new TopicPartition(record.topic(), record.partition()),
                                new OffsetAndMetadata(record.offset() + 1)
                        ));
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }

    private static boolean doBusinessProcess(ConsumerRecord<String, String> record) {
        // 你的业务处理逻辑
        return true;
    }
}

注意事项 #

• 不要无限重试同一条消息，建议设置最大重试次数，超过阈值后将消息写入死信队列人工处理，避免消费链路永久阻塞
• 调整max.poll.interval.ms参数适配你的重试时长，避免单条消息处理/重试时间过长导致消费者被消费组踢出，触发重平衡导致位点重置失效
• 触发seek后必须终止当前批次剩余消息的遍历，否则同批次后面的消息还是会被正常处理，seek操作仅对下一次拉取生效

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