SpringBoot中不使用KafkaAdmin检查Kafka Broker存活状态的方法

以下几种方案可以在无法使用KafkaAdmin的情况下，检查Kafka Broker存活状态或消费者连接状态：

方案1：通过Kafka Consumer实例直接验证连接 #

利用Spring管理的ConsumerFactory创建临时消费者，调用其内置方法查询Topic信息，以此验证与Broker的连接。如果能成功获取到Topic数据，说明Broker存活且消费者可正常连接；若抛出超时、认证等异常，则判定为不健康。

代码实现 #

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.errors.TimeoutException;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Component;

@Component
public class KafkaBrokerHealthIndicator implements HealthIndicator {

    private final ConsumerFactory<String, Object> consumerFactory;
    // 替换为你的应用实际消费的Topic，或专用测试Topic
    private static final String TARGET_TOPIC = "your-consumed-topic";

    public KafkaBrokerHealthIndicator(ConsumerFactory<String, Object> consumerFactory) {
        this.consumerFactory = consumerFactory;
    }

    @Override
    public Health health() {
        try (Consumer<String, Object> consumer = consumerFactory.createConsumer()) {
            // 查询指定Topic的分区信息，验证连接有效性
            consumer.partitionsFor(TARGET_TOPIC);
            return Health.up()
                    .withDetail("kafka-broker-status", "connected")
                    .withDetail("checked-topic", TARGET_TOPIC)
                    .build();
        } catch (TimeoutException e) {
            return Health.down(e)
                    .withDetail("kafka-broker-status", "connection timeout")
                    .build();
        } catch (Exception e) {
            return Health.down(e)
                    .withDetail("kafka-broker-status", "connection failed")
                    .build();
        }
    }
}

注意：需要确保消费者拥有目标Topic的Describe权限，如果没有该权限，可尝试下一种方案。

方案2：监听消费者生命周期事件维护连接状态 #

Spring Kafka会发布消费者生命周期相关事件，我们可以监听这些事件来维护一个全局连接状态标志，健康检查时直接读取该标志即可，无需额外Kafka权限。

步骤1：定义连接状态持有者 #

import org.springframework.stereotype.Component;
import java.util.concurrent.atomic.AtomicBoolean;

@Component
public class KafkaConsumerConnectionStatus {
    private final AtomicBoolean isConnected = new AtomicBoolean(false);

    public boolean isConnected() {
        return isConnected.get();
    }

    public void setConnected(boolean connected) {
        isConnected.set(connected);
    }
}

步骤2：监听消费者事件 #

import org.springframework.context.event.EventListener;
import org.springframework.kafka.event.ConsumerStartedEvent;
import org.springframework.kafka.event.ConsumerStoppedEvent;
import org.springframework.kafka.event.ConsumerFailedToStartEvent;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumerEventListener {

    private final KafkaConsumerConnectionStatus connectionStatus;

    public KafkaConsumerEventListener(KafkaConsumerConnectionStatus connectionStatus) {
        this.connectionStatus = connectionStatus;
    }

    @EventListener
    public void onConsumerStarted(ConsumerStartedEvent event) {
        connectionStatus.setConnected(true);
    }

    @EventListener
    public void onConsumerStopped(ConsumerStoppedEvent event) {
        connectionStatus.setConnected(false);
    }

    @EventListener
    public void onConsumerStartFailed(ConsumerFailedToStartEvent event) {
        connectionStatus.setConnected(false);
    }
}

步骤3：实现健康检查指示器 #

import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumerHealthIndicator implements HealthIndicator {

    private final KafkaConsumerConnectionStatus connectionStatus;

    public KafkaConsumerHealthIndicator(KafkaConsumerConnectionStatus connectionStatus) {
        this.connectionStatus = connectionStatus;
    }

    @Override
    public Health health() {
        if (connectionStatus.isConnected()) {
            return Health.up()
                    .withDetail("consumer-status", "successfully connected to broker")
                    .build();
        } else {
            return Health.down()
                    .withDetail("consumer-status", "not connected to broker")
                    .build();
        }
    }
}

方案3：通过短超时拉取请求验证连接 #

如果上述方案都因权限受限无法使用，可以尝试发起一次短超时的拉取请求，若能正常执行则说明Broker可达，无需额外权限。

代码实现 #

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.errors.WakeupException;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Component;
import java.time.Duration;

@Component
public class KafkaPollHealthIndicator implements HealthIndicator {

    private final ConsumerFactory<String, Object> consumerFactory;

    public KafkaPollHealthIndicator(ConsumerFactory<String, Object> consumerFactory) {
        this.consumerFactory = consumerFactory;
    }

    @Override
    public Health health() {
        try (Consumer<String, Object> consumer = consumerFactory.createConsumer()) {
            // 发起100ms超时的拉取请求，验证连接
            consumer.poll(Duration.ofMillis(100));
            return Health.up()
                    .withDetail("kafka-broker-status", "reachable via poll request")
                    .build();
        } catch (WakeupException e) {
            // 忽略消费者唤醒异常，不影响健康状态判断
            return Health.up().build();
        } catch (Exception e) {
            return Health.down(e)
                    .withDetail("kafka-broker-status", "unreachable via poll request")
                    .build();
        }
    }
}

优化建议：可以为该检查添加缓存逻辑，比如每隔30秒执行一次实际检查，避免每次健康检查都创建临时消费者，减少资源消耗。

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