有用的内容 | Kafka CLI 命令参考

Kafka CLI 命令参考

本章面向 Kafka 3.x ZooKeeper 模式的值班、排障和变更操作。Kafka 的管理 CLI 仍通过 --bootstrap-server 连接 Broker；只有 ZooKeeper 自身诊断才使用 zookeeper-shell.sh 或四字命令。所有会改变数据、位点或副本分布的操作，先在预发或使用 --dry-run 验证。

1. 统一变量与客户端认证

先在当前终端定义路径与集群入口。不要把认证密码、Token 或证书私钥写入命令历史；启用 SASL/SSL 后，将客户端配置放在权限受控的文件中，并通过 --command-config 引用。

# Kafka 安装目录；根据实际部署路径调整。
export KAFKA_HOME=/opt/kafka
# 使用多个稳定 Broker 地址，单个入口故障时客户端仍可建立连接。
export BOOTSTRAP=kafka-1.example.internal:9092,kafka-2.example.internal:9092,kafka-3.example.internal:9092
# 仅在启用 TLS/SASL 时使用；该文件权限应限制为运维账号可读。
export ADMIN_CONFIG=/etc/kafka/admin-client.properties

操作类型	常用命令	风险等级
集群、Topic、消费组查询	`kafka-topics.sh`、`kafka-consumer-groups.sh`	只读
Topic 配置、分区扩容	`kafka-configs.sh`、`kafka-topics.sh --alter`	变更
位点重置、副本重分配、删除 Topic	`kafka-consumer-groups.sh --reset-offsets`、`kafka-reassign-partitions.sh`	高风险

下文未加认证参数的命令假定是隔离实验环境。生产启用认证时，在每条管理命令后补充 --command-config "$ADMIN_CONFIG"，并确认该配置对应最小权限的运维账号。

2. 集群与 Broker 查询

# 列出集群 Topic，快速确认 Broker API 可连接；空输出不表示 Broker 不可用。
$KAFKA_HOME/bin/kafka-topics.sh --list --bootstrap-server "$BOOTSTRAP"

# 查看 Broker 支持的 API 版本，用于核对客户端兼容性和连接是否成功。
$KAFKA_HOME/bin/kafka-broker-api-versions.sh --bootstrap-server "$BOOTSTRAP"

# 查看各 Broker 的日志目录、磁盘使用和未来副本迁移状态。
$KAFKA_HOME/bin/kafka-log-dirs.sh --describe \
  --bootstrap-server "$BOOTSTRAP" --broker-list 1,2,3

# 查看副本不足与不可用分区；两条命令均无输出才是正常状态。
$KAFKA_HOME/bin/kafka-topics.sh --describe --under-replicated-partitions \
  --bootstrap-server "$BOOTSTRAP"
$KAFKA_HOME/bin/kafka-topics.sh --describe --unavailable-partitions \
  --bootstrap-server "$BOOTSTRAP"

kafka-log-dirs.sh 返回的是 Broker 视角的日志目录信息，不等同于操作系统磁盘水位。仍需通过节点监控检查真实磁盘可用空间、IO 延迟和 inode。

3. Topic 全生命周期

# 创建 Topic：显式指定分区、副本和最小 ISR，避免依赖 Broker 默认值。
$KAFKA_HOME/bin/kafka-topics.sh --create --if-not-exists \
  --bootstrap-server "$BOOTSTRAP" --topic orders.created.v1 \
  --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 --config retention.ms=604800000

# 查看指定 Topic 的分区 Leader、副本和 ISR，排查复制或分区倾斜。
$KAFKA_HOME/bin/kafka-topics.sh --describe \
  --bootstrap-server "$BOOTSTRAP" --topic orders.created.v1

# 增加分区数；只能增加，执行前必须评估 Key 路由和顺序语义。
$KAFKA_HOME/bin/kafka-topics.sh --alter \
  --bootstrap-server "$BOOTSTRAP" --topic orders.created.v1 --partitions 18

# 删除 Topic 是异步且高风险操作；先确认 Broker 已允许 delete.topic.enable，并获得业务审批。
$KAFKA_HOME/bin/kafka-topics.sh --delete \
  --bootstrap-server "$BOOTSTRAP" --topic orders.created.v1

Topic 删除可能影响回放、审计和仍在消费的业务。生产中建议先停止生产与消费、导出必要数据和配置、确认保留策略不能满足需求后再执行；不要把“清理积压”误操作为删除 Topic。

4. 生产与消费调试

# 启动交互式生产者；使用 Key 分隔符验证同 Key 消息的分区行为。
$KAFKA_HOME/bin/kafka-console-producer.sh \
  --bootstrap-server "$BOOTSTRAP" --topic orders.created.v1 \
  --property parse.key=true --property key.separator='|'

# 从指定消费者组读取并提交位点；调试时使用独立 group，避免影响生产消费组。
$KAFKA_HOME/bin/kafka-console-consumer.sh \
  --bootstrap-server "$BOOTSTRAP" --topic orders.created.v1 \
  --group ops-debug-orders --from-beginning \
  --property print.key=true --property print.partition=true --property print.offset=true

# 只读取一条消息并退出，适合连通性确认，避免终端持续消费大量历史数据。
$KAFKA_HOME/bin/kafka-console-consumer.sh \
  --bootstrap-server "$BOOTSTRAP" --topic orders.created.v1 \
  --from-beginning --max-messages 1

命令行消费者会真实消费消息并可能提交位点。不要使用生产消费组进行临时调试；需要查看历史消息时新建 ops-debug-* 消费组，完成后按团队保留策略清理。

5. 消费组与位点操作

# 列出消费组，定位需要检查的业务组。
$KAFKA_HOME/bin/kafka-consumer-groups.sh --list \
  --bootstrap-server "$BOOTSTRAP"

# 查看每个分区的已提交位点、日志末端位点、Lag 和消费者成员。
$KAFKA_HOME/bin/kafka-consumer-groups.sh --describe \
  --bootstrap-server "$BOOTSTRAP" --group orders-projection

# 预览重置到最早位点的影响；执行前必须停止该消费组全部实例。
$KAFKA_HOME/bin/kafka-consumer-groups.sh --reset-offsets --dry-run \
  --bootstrap-server "$BOOTSTRAP" --group orders-projection \
  --topic orders.created.v1 --to-earliest

# 审核 dry-run 输出后才执行；该操作可能造成重复处理和下游流量突增。
$KAFKA_HOME/bin/kafka-consumer-groups.sh --reset-offsets --execute \
  --bootstrap-server "$BOOTSTRAP" --group orders-projection \
  --topic orders.created.v1 --to-earliest

位点重置不是修复 Lag 的默认手段。先确认消费者是否仍在处理、Lag 是否增长、消息是否仍在保留期内及下游是否幂等；错误重置可能跳过数据或触发大规模重复写入。

6. 动态配置与 ACL 查询

# 查看 Topic 的动态覆盖配置；没有输出时表示使用 Broker 默认配置。
$KAFKA_HOME/bin/kafka-configs.sh --describe \
  --bootstrap-server "$BOOTSTRAP" \
  --entity-type topics --entity-name orders.created.v1

# 增加 Topic 保留时间；变更后仅影响后续日志清理，不会立即删除已有段文件。
$KAFKA_HOME/bin/kafka-configs.sh --alter \
  --bootstrap-server "$BOOTSTRAP" \
  --entity-type topics --entity-name orders.created.v1 \
  --add-config retention.ms=1209600000

# 查看 ACL；启用 SASL/SSL 时添加 --command-config 并使用专用运维凭据。
$KAFKA_HOME/bin/kafka-acls.sh --list \
  --bootstrap-server "$BOOTSTRAP"

动态配置优先级高于 Broker 默认配置。调整前应先记录原始值；回滚时使用 --delete-config <key> 删除 Topic 级覆盖，而不是在多个 Broker 上手改默认值。

7. 副本重分配

# 根据 Topic 清单和目标 Broker 列表生成候选重分配方案；先审核输出文件。
$KAFKA_HOME/bin/kafka-reassign-partitions.sh --generate \
  --bootstrap-server "$BOOTSTRAP" \
  --topics-to-move-json-file topics.json --broker-list "1,2,3,4"

# 执行审核后的方案；reassignment.json 必须来自 generate 输出并纳入变更记录。
$KAFKA_HOME/bin/kafka-reassign-partitions.sh --execute \
  --bootstrap-server "$BOOTSTRAP" \
  --reassignment-json-file reassignment.json

# 持续查询任务状态；完成前不得下线源或目标 Broker。
$KAFKA_HOME/bin/kafka-reassign-partitions.sh --verify \
  --bootstrap-server "$BOOTSTRAP" \
  --reassignment-json-file reassignment.json

重分配会显著消耗磁盘和网络。必须分批执行，必要时配置复制节流，并监控 ISR、生产延迟和消费 Lag；完成后删除临时节流配置。

8. ZooKeeper 模式诊断

# 查看 ZooKeeper 节点角色；预期一个 leader，其余为 follower。
echo stat | nc -w 3 zk-1.example.internal 2181 | rg 'Mode: (leader|follower)'

# 查看在 ZooKeeper 中注册的 Broker ID；仅用于协调层诊断，不用于 Topic 管理。
$KAFKA_HOME/bin/zookeeper-shell.sh zk-1.example.internal:2181 \
  ls /kafka/brokers/ids

# 查看 Controller 所在 Broker ID；Controller 频繁变化时检查 ZooKeeper 会话、网络和 GC。
$KAFKA_HOME/bin/zookeeper-shell.sh zk-1.example.internal:2181 \
  get /kafka/controller

ZooKeeper Shell 命令可能返回大量内部元数据。只读查询可用于故障分析，但禁止直接删除 /kafka 下的 znode；这会破坏仍在运行的 Kafka 集群元数据。

9. 值班前检查表

确认 $BOOTSTRAP 指向正确环境，避免在错误集群执行变更。
对高风险命令先使用 --dry-run、--generate 或只读 --describe。
记录变更前的 Topic 配置、ISR、消费组位点和 Broker 磁盘状态。
执行过程中持续观察 Under Replicated Partitions、生产错误率和 Consumer Lag。
完成后验证业务读写、ISR 恢复和告警回归，再关闭变更单。