如何用Spring Boot检查Kafka集群连通性与主题可用性状态

一、集群连通性检查实现 #

验证Kafka集群连通性的核心是尝试与集群建立连接并获取元数据。可以通过AdminClient的listTopics()方法，配合超时控制判断连通状态：

• 若能在指定时间内获取到集群元数据，说明连通正常；
• 若抛出超时、连接失败等异常，则判定集群不可达。

注：如果方法参数bootstrapServer和已注入的AdminClient配置地址不一致，需要临时创建新的AdminClient实例检查指定地址；若地址一致，直接复用现有实例即可。

二、主题存在性检查实现 #

利用AdminClient.describeTopics()方法查询指定主题信息：

• 查询成功且返回有效主题信息，说明主题可用；
• 若抛出主题不存在、超时或权限异常，返回false。

完整代码实现 #

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.springframework.stereotype.Component;
import lombok.AllArgsConstructor;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

@Component
@AllArgsConstructor
public class KafkaAdministrator {

    private final AdminClient adminClient;
    // 可根据业务场景调整超时时间
    private static final int CHECK_TIMEOUT_SECONDS = 10;

    public boolean isKafkaClusterExists(String bootstrapServer) {
        AdminClient targetClient = adminClient;
        // 若传入地址与现有配置不一致，创建临时AdminClient
        if (!adminClient.config().get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG).equals(bootstrapServer)) {
            Properties props = new Properties();
            props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
            targetClient = AdminClient.create(props);
        }

        try {
            ListTopicsOptions options = new ListTopicsOptions();
            options.timeoutMs((int) TimeUnit.SECONDS.toMillis(CHECK_TIMEOUT_SECONDS));
            ListTopicsResult result = targetClient.listTopics(options);
            // 尝试获取主题列表，能拿到结果则说明集群连通
            result.names().get(CHECK_TIMEOUT_SECONDS, TimeUnit.SECONDS);
            return true;
        } catch (Exception e) {
            // 捕获所有连接类异常：超时、连接失败、权限问题等
            return false;
        } finally {
            // 关闭临时创建的AdminClient，避免资源泄漏
            if (targetClient != adminClient) {
                targetClient.close();
            }
        }
    }

    public boolean isTopicExists(String topic) {
        try {
            DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(topic));
            // 超时时间内获取到主题信息则判定存在
            result.all().get(CHECK_TIMEOUT_SECONDS, TimeUnit.SECONDS);
            return true;
        } catch (TimeoutException e) {
            // 请求超时，判定主题不可用或集群不通
            return false;
        } catch (TopicAuthorizationException e) {
            // 无权限访问主题，判定不可用
            return false;
        } catch (Exception e) {
            // 其他异常（如主题不存在的封装异常），返回false
            return false;
        }
    }
}

关键注意事项 #

• 超时控制：必须设置合理超时时间，避免方法长时间阻塞；
• 资源释放：临时创建的AdminClient要在finally块关闭，防止资源泄漏；
• 异常覆盖：捕获Kafka操作的全量异常，统一返回false，匹配方法布尔值返回逻辑。

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