如何在Kafka消费者端感知Broker不可用并触发告警？

针对你遇到的Broker不可用需要触发告警的需求，你当前使用的NonResponsiveConsumerEvent主要用于捕获消费者线程长时间未执行poll操作的场景，并不适用于Broker连接失败的情况。以下是更精准的实现方案：

一、监听容器故障事件 #

通过Spring Kafka发布的容器事件，可以直接捕获Broker不可用导致的容器启动失败或运行中断：

修改消费者服务类 #

@Service
public class Consumer {
    private static final Logger logger = LoggerFactory.getLogger(Consumer.class);

    @KafkaListener(topics = "topic", groupId = "groupName")
    public void consumer(String message) {
        logger.info(message);
    }

    // 监听Kafka容器启动/运行失败事件（覆盖Broker不可用导致的故障）
    @EventListener
    public void handleContainerFailure(ListenerContainerFailedEvent event) {
        // 触发告警逻辑（如调用告警接口、发送通知等）
        logger.error("Kafka容器故障，无法连接Broker: {}", event.getThrowable().getMessage(), event.getThrowable());
        // TODO: 此处添加你的告警代码
    }

    // 监听消费者启动失败事件（针对初始化阶段的Broker连接问题）
    @EventListener
    public void handleConsumerStartFailure(ConsumerFailedToStartEvent event) {
        logger.error("消费者启动失败，Broker不可用: {}", event.getThrowable().getMessage(), event.getThrowable());
        // TODO: 此处添加你的告警代码
    }
}

二、配置自定义错误处理器 #

在消费过程中如果出现Broker连接中断的异常，可通过自定义ErrorHandler捕获并触发告警：

修改Kafka配置类 #

@EnableKafka
@Configuration
public class KafkaConfig {

    @Bean
    public ConsumerFactory<Integer, String> createConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "groupName");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 配置重连间隔，避免频繁重试同时保证异常能被及时捕获
        props.put(ConsumerConfig.RECONNECT_BACKOFF_MS_CONFIG, 5000);
        props.put(ConsumerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG, 30000);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(createConsumerFactory());
        
        // 配置自定义错误处理器，捕获消费过程中的Broker连接异常
        factory.setErrorHandler((thrownException, data) -> {
            boolean isBrokerUnavailable = thrownException instanceof ConnectException 
                    || (thrownException.getCause() != null && thrownException.getCause() instanceof ConnectException);
            if (isBrokerUnavailable) {
                logger.error("消费过程中无法连接Kafka Broker，触发告警", thrownException);
                // TODO: 此处添加你的告警代码
            }
        });

        factory.getContainerProperties().setIdleEventInterval(5L);
        factory.getContainerProperties().setNoPollThreshold(1L);
        factory.getContainerProperties().setMonitorInterval(1);
        return factory;
    }

    @Bean
    public Consumer consumer() {
        return new Consumer();
    }
}

三、关键说明 #

• ListenerContainerFailedEvent：覆盖容器启动失败、运行中因Broker不可用导致的中断场景，是捕获Broker故障的核心事件。
• ConsumerFailedToStartEvent：针对消费者初始化阶段无法连接Broker的情况，补充覆盖启动时的异常。
• 自定义ErrorHandler：处理消费循环中出现的连接异常，确保运行时的连接中断也能触发告警。
• 重连配置参数：通过RECONNECT_BACKOFF_MS_CONFIG和RECONNECT_BACKOFF_MAX_MS_CONFIG控制重连间隔，避免无效重试同时保证异常能被及时捕捉。

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