Spring Kafka 1.2.x消费者能否实现SeekToCurrentErrorHandler等效行为？

没问题，在Spring Kafka 1.2.x里完全可以实现和SeekToCurrentErrorHandler一样的效果——不用停止容器，就能让处理失败的消息重新投递消费。我先帮你分析下当前代码的问题，再给出具体的修改方案：

你当前代码的核心问题 #

你现在尝试用seek回到失败消息的偏移，但有两个关键漏洞：

1. 没有维护分区的失败偏移状态：当容器完成一轮消息拉取后，内部的消费位置可能已经前进，导致后续重复拉取分区的最后一条消息
2. ThreadLocal<ConsumerSeekCallback>的使用没有结合分区分配事件：当分区重新分配时，无法正确恢复之前的偏移位置

解决方案：修改监听器+完善偏移管理 #

我们需要通过线程安全的映射记录每个分区的失败偏移，在处理失败、分区分配、容器空闲时分别做对应的seek操作，确保失败消息能被重新拉取。

修改后的监听器代码

@Component
public class Listener implements ConsumerSeekAware {
    private static final Logger logger = LoggerFactory.getLogger(Listener.class);
    // 线程安全存储：分区 -> 需要重新消费的偏移
    private final ConcurrentMap<TopicPartition, Long> failedOffsets = new ConcurrentHashMap<>();
    // 每个消费者线程对应的Seek回调
    private final ThreadLocal<ConsumerSeekCallback> seekCallback = new ThreadLocal<>();

    @KafkaListener(topics = "my-topic", containerFactory = "kafkaManualAckListenerContainerFactory")
    public void listen1(ConsumerRecord<String, String> consumerRecord, Acknowledgment ack) throws MyCustomException {
        TopicPartition tp = new TopicPartition(consumerRecord.topic(), consumerRecord.partition());
        long currentOffset = consumerRecord.offset();
        logger.info("received: key - {} value - {} offset - {}", consumerRecord.key(), consumerRecord.value(), currentOffset);

        boolean shouldCommit = false;
        try {
            // 这里替换成你的实际业务处理逻辑
            // ...

            if (shouldCommit) {
                ack.acknowledge();
                // 处理成功，移除该分区的失败偏移记录
                failedOffsets.remove(tp);
            }
        } catch (Exception e) {
            logger.error("处理消息失败，准备重新消费", e);
            // 记录当前失败消息的偏移
            failedOffsets.put(tp, currentOffset);
            // 立即seek回当前偏移，确保下一次poll能拉取这条消息
            seekCallback.get().seek(tp.topic(), tp.partition(), currentOffset);
            // 注意：不要调用ack，确保偏移不会被提交
        }
    }

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        logger.info("registerSeekCallback called..");
        this.seekCallback.set(callback);
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        logger.info("onPartitionsAssigned called..");
        // 分区分配时，恢复之前记录的失败偏移
        assignments.keySet().forEach(tp -> {
            Long failedOffset = failedOffsets.get(tp);
            if (failedOffset != null) {
                logger.info("恢复分区 {} 的消费偏移到 {}", tp, failedOffset);
                callback.seek(tp.topic(), tp.partition(), failedOffset);
            }
        });
        // 更新当前线程的Seek回调
        this.seekCallback.set(callback);
    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        logger.info("onIdleContainer called..");
        // 容器空闲时（没有新消息），重新seek到失败偏移，解决你遇到的"重复拉取最后一条消息"问题
        assignments.keySet().forEach(tp -> {
            Long failedOffset = failedOffsets.get(tp);
            if (failedOffset != null) {
                logger.info("容器空闲，重新定位分区 {} 到偏移 {}", tp, failedOffset);
                callback.seek(tp.topic(), tp.partition(), failedOffset);
            }
        });
    }
}

确认容器配置正确

确保你的容器配置保持以下设置，关闭自动提交，使用手动立即确认模式：

// 关闭错误时自动ack
factory.getContainerProperties().setAckOnError(false);
// 设置手动立即确认模式
factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
// 强制关闭消费者自动提交偏移
factory.getConsumerProperties().put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

关键逻辑说明 #

1. 失败偏移持久化：用ConcurrentHashMap记录每个分区的失败偏移，即使消费者重启或分区重分配，也能恢复到正确的消费位置
2. 多场景seek触发：
   • 消息处理失败时：立即seek回当前偏移，确保下一次poll优先拉取这条失败消息
   • 分区分配时：恢复之前的失败偏移，避免分区切换后丢失未处理的消息
   • 容器空闲时：重新seek到失败偏移，解决你遇到的"接收完所有消息后持续拉取最后一条"的问题——当容器没有新消息时，主动重置消费位置到失败偏移，触发重新拉取
3. 偏移提交控制：只有处理成功时才调用ack.acknowledge()，并清除失败偏移记录；失败时绝不提交偏移，确保消息不会被标记为已消费

这样修改后，就能实现和SeekToCurrentErrorHandler完全一致的行为：失败消息自动重新投递，无需停止容器，也不会出现重复拉取最后一条消息的问题。

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