Spring-Kafka遇异常时停止消费后续消息的实现方案咨询

使用Spring-Kafka 2.8.1（基于Spring Boot 2.7）开发Kafka消费者，需求为：当监听器处理消息遇到异常时，立即停止处理所有后续消息，等待异常条件解决（需手动干预）后再恢复消费。

当前已配置手动确认模式的消费者容器，在监听器处理包含"BLENDER"的消息时主动抛出RuntimeException，但测试发现非BLENDER相关的消息仍会被正常消费，不符合预期。

现有配置代码 #

@Bean("consumerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, ? extends SpecificRecordBase>
kafkaListenerContainerFactory() {

    String concurrency = consumer == null ?  "1" : consumer.getConcurrency();
    ConcurrentKafkaListenerContainerFactory<String, ? extends SpecificRecordBase> factory =
            new ConcurrentKafkaListenerContainerFactory<>();

    factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
    factory.setConcurrency(Integer.parseInt(concurrency));
    factory.setCommonErrorHandler(getDefaultErrorHandler());
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
    log.info("Kafka Consumer Created Successfully");
    return factory;
}

@Bean
public DefaultErrorHandler getDefaultErrorHandler() {
    DefaultErrorHandler defaultErrorHandler = new DefaultErrorHandler();
    defaultErrorHandler.setAckAfterHandle(false);
    return defaultErrorHandler;
}

public Map<String, Object> consumerConfigs() {

    ConsumerConfiguration config = consumer == null ? new ConsumerConfiguration() : consumer;
    Map<String, Object> consumerProperties = new HashMap<>(kafkaCommonConfigs());

    consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);

    if (StringUtils.isNotBlank(config.getMaxPollInterval())) {
        consumerProperties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,
                Integer.parseInt(config.getMaxPollInterval()));
    }

    if (StringUtils.isNotBlank(config.getSessionTimeout())) {
        consumerProperties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,
                Integer.parseInt(config.getSessionTimeout()));
    }
    consumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, EARLIEST);
    consumerProperties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, READ_COMMITTED);
    consumerProperties.put(SPECIFIC_AVRO_READER_CONFIG, true);

    return consumerProperties;
}

现有监听器代码 #

@KafkaListener(topics = "#{topicName}",
            groupId = "#{consumerGroupId}",
            containerFactory = "consumerContainerFactory")
public void listenProgramNumberMessages(ConsumerRecord<String, Event> record,
                                        Acknowledgment acknowledgment) {
    try {
        var event = record.value();
        log.info("Received message: {}", event);
        if(isProcessingAllowed(event)){
            log.info("Consumed message: {}" , marketChannelCustomerChoiceEvent);
            if(event.getCustomerChoice().getAbbreviatedDescription()
                    .contains("BLENDER")){
                throw new RuntimeException();
            }

            listenerService.doSomething();
            acknowledgment.acknowledge();
        }
    } catch (Exception e) {
        log.error("Exception occurred :", e);
        acknowledgment.nack(Duration.ofMillis(1000));
    }
}

要实现异常触发后暂停所有消费、手动恢复的逻辑，核心是通过KafkaListenerEndpointRegistry管理消费者容器，在异常发生时暂停容器，并提供手动恢复的入口。

步骤1：为监听器指定唯一ID并注入容器注册表 #

给@KafkaListener添加唯一ID，方便后续定位容器；同时注入KafkaListenerEndpointRegistry用于管理容器状态：

@Autowired
private KafkaListenerEndpointRegistry listenerRegistry;

@KafkaListener(id = "blenderEventConsumer", // 添加唯一ID
               topics = "#{topicName}",
               groupId = "#{consumerGroupId}",
               containerFactory = "consumerContainerFactory")
public void listenProgramNumberMessages(ConsumerRecord<String, Event> record,
                                        Acknowledgment acknowledgment) {
    // 原有逻辑...
}

步骤2：修改异常处理逻辑，暂停消费者容器 #

移除原有的acknowledgment.nack()，改为在捕获异常时暂停整个容器：

catch (Exception e) {
    log.error("Exception occurred, pausing all consumer processing:", e);
    // 获取当前监听器对应的容器
    MessageListenerContainer container = listenerRegistry.getListenerContainer("blenderEventConsumer");
    if (container != null) {
        container.pause(); // 暂停所有并发的消费者线程
    }
    // 不提交偏移量，恢复后从当前未确认的消息开始处理
}

步骤3：实现手动恢复消费的入口 #

通过REST接口提供手动恢复的能力，异常解决后调用该接口恢复消费：

@RestController
@RequestMapping("/kafka/consumer")
public class KafkaConsumerControlController {

    @Autowired
    private KafkaListenerEndpointRegistry listenerRegistry;

    @PostMapping("/resume")
    public String resumeBlenderConsumer() {
        MessageListenerContainer container = listenerRegistry.getListenerContainer("blenderEventConsumer");
        if (container != null) {
            if (container.isPaused()) {
                container.resume();
                return "Consumer container resumed successfully";
            } else {
                return "Consumer container is not paused";
            }
        }
        return "Consumer container not found";
    }
}

关键说明 #

• 容器暂停后，Kafka消费者会停止拉取新消息，直到调用resume()方法。
• 未确认的偏移量不会提交，恢复消费后会从当前卡住的消息开始重新处理。
• 如果配置了并发（concurrency > 1），pause()会暂停该监听器对应的所有并发消费者实例，确保所有消息处理都停止。
• 原有的DefaultErrorHandler配置无需修改，因为异常已在监听器内部捕获并处理，不会触发ErrorHandler的重试逻辑。

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