如何识别Kafka集群运行模式：Kraft模式还是ZooKeeper模式？

我们拥有数十个Kafka集群测试环境，部分采用Kraft模式，多数为传统ZooKeeper部署的Kafka集群。现需基于python/perl/bash编写的冒烟测试脚本，自动识别各集群的运行模式是Kraft还是非Kraft。以下是已测试但效果不佳的方案：

已尝试的不理想方案 #

1. 通过Kafka PID识别 #

使用命令：

ps -ef | grep ` netstat -tulpn | grep "9092" | awk '{print $NF}' | sed s'/// /g' | awk '{print $1}' ` | grep -v grep | grep -i kraft

问题：仅当Kafka二进制文件路径包含"Kraft"关键字时才能过滤，若路径无该关键字则失效。

2. 使用Kafka CLI工具kafka-metadata-quorum.sh #

• Kraft模式下执行结果：
  运行命令bash kafka-metadata-quorum.sh --bootstrap-server $(hostname):9092 describe --status，返回正常集群元数据：

  ClusterId:              uuZQoN9-TGmVU3lKDFCtww
  LeaderId:               2001
  LeaderEpoch:            975
  HighWatermark:          224514
  MaxFollowerLag:         0
  MaxFollowerLagTimeMs:   0
  CurrentVoters:          [2001,2002,2003]
  CurrentObservers:       [1010,1001,1002,1003]
  

• ZooKeeper模式下执行结果：
  运行相同命令会抛出UnsupportedVersionException错误：

  kafka-metadata-quorum.sh  --bootstrap-server $(hostname):9092 describe --status
  org.apache.kafka.common.errors.UnsupportedVersionException: The broker does not support DESCRIBE_QUORUM
  java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnsupportedVersionException: The broker does not support DESCRIBE_QUORUM
          at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
          at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
          at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
          at org.apache.kafka.tools.MetadataQuorumCommand.handleDescribeStatus(MetadataQuorumCommand.java:208)
          at org.apache.kafka.tools.MetadataQuorumCommand.execute(MetadataQuorumCommand.java:108)
          at org.apache.kafka.tools.MetadataQuorumCommand.mainNoExit(MetadataQuorumCommand.java:61)
          at org.apache.kafka.tools.MetadataQuorumCommand.main(MetadataQuorumCommand.java:56)
  Caused by: org.apache.kafka.common.errors.UnsupportedVersionException: The broker does not support DESCRIBE_QUORUM
  

问题：虽然能区分模式，但依赖CLI工具的可用性，且错误输出的捕获逻辑在脚本中处理相对繁琐。

3. 创建标识文件 #

在/var/lib/kafka/kraft路径下创建标识文件，通过检查该文件是否存在判断模式。
问题：标识文件可能被用户误删，可靠性无法保障。

4. 识别Controller PID #

通过检测Controller进程是否运行判定为Kraft模式。
问题：仅适用于Broker与Controller同主机的组合模式，对Controller独立部署的集群无效；且Controller宕机时会出现误判。

推荐的更优方案 #

方案一：检查Kafka配置文件核心参数 #

直接读取Kafka的核心配置文件（通常为server.properties），通过关键配置项判断：

• 若配置了process.roles（值包含broker/controller）且未配置zookeeper.connect，则为Kraft模式
• 若配置了zookeeper.connect且未配置process.roles，则为ZooKeeper模式

示例bash脚本片段：

CONFIG_FILE="/path/to/kafka/config/server.properties"

# 提取配置参数
PROCESS_ROLES=$(grep -E "^process.roles" $CONFIG_FILE | awk -F= '{print $2}' | tr -d ' ')
ZOOKEEPER_CONNECT=$(grep -E "^zookeeper.connect" $CONFIG_FILE | awk -F= '{print $2}' | tr -d ' ')

if [[ -n "$PROCESS_ROLES" && -z "$ZOOKEEPER_CONNECT" ]]; then
    echo "Kraft模式"
elif [[ -n "$ZOOKEEPER_CONNECT" && -z "$PROCESS_ROLES" ]]; then
    echo "ZooKeeper模式"
else
    echo "无法识别的模式（可能存在混合配置）"
fi

优势：基于集群核心配置判断，准确性高，不受进程状态、文件删除等因素影响，脚本实现简单。

方案二：通过Kafka AdminClient API查询 #

使用Python的kafka-python库或Java AdminClient调用API，远程查询集群配置信息：

• Kraft模式下，Broker配置会包含process.roles且无zookeeper.connect
• ZooKeeper模式下，Broker配置会包含zookeeper.connect且无process.roles

示例Python代码片段：

from kafka import KafkaAdminClient

def detect_kafka_mode(bootstrap_servers):
    try:
        admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)
        # 查询Broker 0的配置信息
        configs = admin_client.describe_configs(resources=[('broker', '0')])['broker-0']
        process_roles = configs.get('process.roles', None)
        zk_connect = configs.get('zookeeper.connect', None)
        
        if process_roles and not zk_connect:
            return "Kraft模式"
        elif zk_connect and not process_roles:
            return "ZooKeeper模式"
        else:
            return "无法识别的模式"
    except Exception as e:
        return f"检测失败: {str(e)}"

# 使用示例
print(detect_kafka_mode("localhost:9092"))

优势：无需直接访问配置文件，适合远程集群检测，准确性高，易于集成到Python编写的冒烟测试脚本中。

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