Spring Kafka禁用自动提交时DefaultErrorHandler仍提交偏移量问题

问题原因 #

在Spring Kafka 2.8.11版本中，FailedBatchProcessor（DefaultErrorHandler的父类）的seekOrRecover逻辑设计上，当失败记录的退避次数耗尽并被跳过之后，会默认尝试提交已成功处理部分的记录偏移量——目的是确保这部分记录不会被重复处理。

该逻辑没有检查enable.auto.commit=false或者group.id=null的配置，因为框架默认消费者会配置有效的group.id，即便使用手动Ack模式，错误处理流程也会尝试提交已处理记录的偏移量，这就导致你的场景下因group.id缺失而抛出InvalidGroupIdException。

解决方案 #

方案一：自定义ErrorHandler跳过提交逻辑 #

继承DefaultErrorHandler，重写seekOrRecover方法，移除其中的偏移量提交步骤，只保留跳过失败记录的逻辑：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.listener.SeekUtils;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.util.backoff.FixedBackOff;
import java.util.List;

public class NoCommitDefaultErrorHandler extends DefaultErrorHandler {

    // 根据你的需求配置退避策略，示例为3次退避，每次间隔1秒
    public NoCommitDefaultErrorHandler() {
        super(new FixedBackOff(1000L, 3));
    }

    @Override
    protected void seekOrRecover(Consumer<?, ?> consumer, List<ConsumerRecord<?, ?>> records,
                                 RuntimeException exception, boolean recoverable, ContainerProperties containerProperties) {
        // 仅执行跳过失败记录的逻辑，不提交偏移量
        SeekUtils.seekOrRecover(consumer, records, exception, recoverable, containerProperties, getRecoveryStrategy());
        // 移除原逻辑中的commitManualAck调用，避免触发偏移量提交
    }
}

在配置容器时使用这个自定义ErrorHandler：

@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConsumerFactory<?, ?> consumerFactory) {
    ConcurrentKafkaListenerContainerFactory<?, ?> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setBatchListener(true);
    // 替换为自定义的ErrorHandler
    factory.setErrorHandler(new NoCommitDefaultErrorHandler());
    // 保持手动Ack模式配置
    factory.getContainerProperties().setAckMode(AckMode.MANUAL);
    return factory;
}

方案二：配置临时group.id（备选） #

如果可以接受配置一个临时的group.id，可以设置一个无实际业务意义的group名称，同时保持enable.auto.commit=false和auto.offset.reset=earliest。这样即使错误处理流程尝试提交偏移量，也不会抛出异常，而服务重启后依然会从最早的记录开始消费，完全符合日志压缩主题的消费需求。

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