命令行工具：生产、消费、查 lag

bin/ 下二十多个脚本，日常只用八个。全部用 --bootstrap-server，--zookeeper 参数在 3.x 已废弃。

场景

排障时手边只有一台跳板机和 bin/ 目录。这些命令要能背出来。

一、kafka-topics.sh

bash

B=10.0.0.31:9092
K=/opt/kafka/bin

# 列表
$K/kafka-topics.sh --bootstrap-server $B --list

# 详情
$K/kafka-topics.sh --bootstrap-server $B --describe --topic order

# 只看异常
$K/kafka-topics.sh --bootstrap-server $B --describe --under-replicated-partitions
$K/kafka-topics.sh --bootstrap-server $B --describe --unavailable-partitions

# 按 topic 过滤列表（3.x 无内置，用 grep）
$K/kafka-topics.sh --bootstrap-server $B --list | grep '^order'

# 删除（生产是异步的，先确认 delete.topic.enable=true）
$K/kafka-topics.sh --bootstrap-server $B --delete --topic order

text

order
order-dlq
smoke
__consumer_offsets

Topic: order	TopicId: 7cQ3nR8vW1xY4zA6bC9dEf	PartitionCount: 6	ReplicationFactor: 3	Configs: min.insync.replicas=2
	Topic: order	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2

--delete 没有二次确认

命令直接返回，删除是后台进行的。topic 一删，数据不可恢复（除非开了 tiered storage）。生产执行前先 --describe 确认名字没打错，并且问清楚消费方。

二、kafka-console-producer.sh

bash

# 不带 key
printf 'msg1\nmsg2\n' | $K/kafka-console-producer.sh --bootstrap-server $B --topic smoke

# 带 key：用 : 分隔，并指定解析器
printf 'order-1:{"id":1}\norder-2:{"id":2}\n' | \
  $K/kafka-console-producer.sh --bootstrap-server $B --topic order \
  --property parse.key=true --property key.separator=:

# 从文件灌（大文件用它）
$K/kafka-console-producer.sh --bootstrap-server $B --topic order < /tmp/orders.txt

# 自定义分区器相关：直接指定分区（控制台工具不支持，需用 kafka-verifiable-producer 或客户端）

text

>{"id":1}
>{"id":2}
>

用 key 时 : 后面才是 value，: 之前是 key。如果 value 里本身有 :，要换一个 key.separator（如 |），否则 key 会被切错。

验证 key 是否写进去了，消费时打印 key：

bash

$K/kafka-console-consumer.sh --bootstrap-server $B --topic order \
  --from-beginning --max-messages 2 --property print.key=true --property print.partition=true

text

Partition:3	order-1	{"id":1}
Partition:1	order-2	{"id":2}

三、kafka-console-consumer.sh

bash

# 从头读（临时 group，读完即退出）
$K/kafka-console-consumer.sh --bootstrap-server $B --topic order \
  --from-beginning --max-messages 10 \
  --property print.key=true --property print.partition=true --property print.timestamp=true

# 只读某个分区（排障最常用）
$K/kafka-console-consumer.sh --bootstrap-server $B --topic order --partition 3 --offset 0 --max-messages 5

# 从指定时间点读（需要先查 offset）
$K/kafka-console-consumer.sh --bootstrap-server $B --topic order \
  --partition 0 --offset 1200 --max-messages 5

# 带 group 长期跑（会提交位移，注意别污染业务 group）
$K/kafka-console-consumer.sh --bootstrap-server $B --topic order --group debug-tmp

text

CreateTime:1761549000000	Partition:0	order-1	{"id":1}
CreateTime:1761549001200	Partition:1	order-2	{"id":2}
Processed a total of 2 messages

--from-beginning 只对没有位移的 group 生效

如果用了 --group，而这个 group 已经提交过 offset，--from-beginning 会被忽略——它从已提交位置继续。要看历史必须：

换一个新的 group 名，或
用 --partition X --offset 0 显式指定，或
用 kafka-consumer-groups.sh --reset-offsets 重置（会影响该 group 全局）。

四、kafka-consumer-groups.sh

这是查积压的主命令。

bash

# 所有 group
$K/kafka-consumer-groups.sh --bootstrap-server $B --list

# 某个 group 的每个分区进度
$K/kafka-consumer-groups.sh --bootstrap-server $B --describe --group order-group

text

Consumer group 'order-group' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG     CONSUMER-ID  HOST  CLIENT-ID
order-group     order           0          1204            1204            0       -            -     -
order-group     order           1          998             1200            202     -            -     -
order-group     order           2          1355            1355            0       -            -     -

字段：

字段	含义
`CURRENT-OFFSET`	已提交的消费位置
`LOG-END-OFFSET`	分区末尾
`LAG`	差值，即积压条数
`CONSUMER-ID`	在线消费者；全是 `-` 说明没有消费者在跑

has no active members + CONSUMER-ID 全是 -，通常意味着消费服务挂了。这是排查积压第一个要看的。

读日志时注意 LAG 可能为负：read_committed 隔离级别下，LOG-END-OFFSET 用的是 LSO（last stable offset），事务未提交时 LAG 计算会有点绕。

bash

# 所有 group 的总积压（快速巡检）
$K/kafka-consumer-groups.sh --bootstrap-server $B --all-groups --describe \
  | awk 'NR>1 && $6 ~ /^[0-9]+$/ {sum[$1]+=$6} END {for (g in sum) print sum[g], g}' | sort -rn | head

text

182400 order-group
     0 report-group

重置位移

必须先停掉所有消费者，否则不会生效或会互相覆盖。

bash

# 重置到最早
$K/kafka-consumer-groups.sh --bootstrap-server $B --group order-group \
  --topic order --reset-offsets --to-earliest --execute

# 重置到最新（相当于丢弃积压）
$K/kafka-consumer-groups.sh --bootstrap-server $B --group order-group \
  --topic order --reset-offsets --to-latest --execute

# 回到 2 小时前
$K/kafka-consumer-groups.sh --bootstrap-server $B --group order-group \
  --topic order --reset-offsets --to-datetime 2026-10-27T17:00:00.000 --execute

# 往前回退 1000 条
$K/kafka-consumer-groups.sh --bootstrap-server $B --group order-group \
  --topic order --reset-offsets --shift-by -1000 --execute

text

GROUP           TOPIC    PARTITION  NEW-OFFSET
order-group     order    0          1000
order-group     order    1          0
order-group     order    2          1355

重置位移前先确认幂等

--to-earliest 会让该 group 重新消费所有还没被 retention 删掉的消息。如果消费逻辑不幂等，下游会收到大量重复数据。

正确做法：先 --dry-run 看 NEW-OFFSET，再执行；数据敏感的组先在同名测试组演练。

删除 group

bash

$K/kafka-consumer-groups.sh --bootstrap-server $B --delete --group debug-tmp

只能删没有活跃成员的 group。

五、kafka-get-offsets.sh

查某个时间点的 offset，比重置位移更精确。

bash

# 最新位移
$K/kafka-get-offsets.sh --bootstrap-server $B --topic order --time -1

# 最早位移
$K/kafka-get-offsets.sh --bootstrap-server $B --topic order --time -2

# 指定时间戳（毫秒）
$K/kafka-get-offsets.sh --bootstrap-server $B --topic order --time 1761523200000

# 按时间再往前推 2 小时
$K/kafka-get-offsets.sh --bootstrap-server $B --topic order --time -3

text

order:0:1204
order:1:1200
order:2:1355
order:3:1201
order:4:1198
order:5:1205

--time 的特殊值：

值	含义
`-1`	最新（latest）
`-2`	最早（earliest）
`-3`	当前时间 - 2 小时。注意不是 -2 小时，是"比现在早 2 小时"这个语义的特殊值
正整数	Unix 毫秒时间戳
`-4`	每个分区最大时间戳

分区总消息数 = 各分区 latest 之和减去 earliest 之和。

六、kafka-log-dirs.sh

bash

$K/kafka-log-dirs.sh --bootstrap-server $B --describe --topic-list order | python3 -m json.tool | head -40

json

{
    "version": 1,
    "brokers": [
        {
            "broker": 1,
            "logDirs": [
                {
                    "logDir": "/data/kafka/logs",
                    "error": null,
                    "partitions": [
                        { "partition": "order-0", "size": 2097152000, "offsetLag": 0, "isFuture": false },
                        { "partition": "order-2", "size": 2147483648, "offsetLag": 0, "isFuture": false }
                    ]
                }
            ]
        }
    ]
}

offsetLag 指的是这个 broker 上作为 follower 的副本落后多少。isFuture 为 true 表示这个副本正在被调整。

七、kafka-dump-log.sh

直接看磁盘上的日志文件，能确认消息真的写进去了、格式对不对。

bash

$K/kafka-dump-log.sh --files /data/kafka/logs/order-0/00000000000000000000.log --print-data-log | head -3

text

Dumping /data/kafka/logs/order-0/00000000000000000000.log
Starting offset: 0
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 0 CreateTime: 1761549000000 size: 78 magic: 2 compresscodec: none crc: 3612345678 isvalid: true
payload: {"id":1}

字段	用途
`baseOffset` / `lastOffset`	这一批消息的范围
`count`	批次里几条
`partitionLeaderEpoch`	写入时的 leader 纪元，排障脑裂用
`isTransactional`	是否事务消息
`compresscodec`	压缩算法（`none`/`gzip`/`snappy`/`lz4`/`zstd`）
`crc` / `isvalid`	校验和，`isvalid: false` 说明文件损坏

大文件不要全量 dump

--print-data-log 会把消息体打出来，1GB 的 segment 会刷屏几个小时。加 head，或用 --print-data-log | grep -c payload 统计。

八、kafka-leader-election.sh

bash

# 所有分区把 preferred leader 抢回来（扩缩容后必做）
$K/kafka-leader-election.sh --bootstrap-server $B --election-type preferred --all-topic-partitions

# 单个分区
$K/kafka-leader-election.sh --bootstrap-server $B --election-type preferred --topic order --partition 0

# 从 JSON 文件批量
$K/kafka-leader-election.sh --bootstrap-server $B --election-type preferred --path-to-json-file /tmp/partitions.json

text

Successfully completed preferred leader election for 6 partitions.

「preferred leader」是 Replicas 列表里的第一个。正常情况 leader 应该长期是它。重启或抖动后 leader 可能漂移，定期做一次 preferred 选举把负载拉回均衡。

九、kafka-reassign-partitions.sh

扩 broker 或修倾斜时用。三步走，第二步才是真执行。

1. 准备 topic 列表

bash

cat > /tmp/topics.json <<'EOF'
{"topics":[{"topic":"order"}],"version":1}
EOF

2. 生成方案

bash

$K/kafka-reassign-partitions.sh --bootstrap-server $B \
  --topics-to-move-json-file /tmp/topics.json \
  --broker-list "1,2,3,4" --generate

text

Current partition replica assignment
{"version":1,"partitions":[{"topic":"order","partition":0,"replicas":[3,1,2],"log_dirs":["any","any","any"]},...]}

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"order","partition":0,"replicas":[1,2,4],"log_dirs":["any","any","any"]},...]}

把 Proposed 那段存成文件：

bash

# 手工复制到 /tmp/reassign.json，或在脚本里截取
$K/kafka-reassign-partitions.sh --bootstrap-server $B \
  --topics-to-move-json-file /tmp/topics.json \
  --broker-list "1,2,3,4" --generate > /tmp/plan.txt

3. 执行与校验

bash

$K/kafka-reassign-partitions.sh --bootstrap-server $B \
  --reassignment-json-file /tmp/reassign.json --execute \
  --throttle 52428800

text

Current partition replica assignment
...
Successfully started partition reassignments for order-0, order-1, ...

--throttle 52428800 是 50MB/s 的限速，必须加，否则迁移会打满网络，影响正常读写。

bash

$K/kafka-reassign-partitions.sh --bootstrap-server $B \
  --reassignment-json-file /tmp/reassign.json --verify

text

Status of partition reassignment:
Reassignment of partition order-0 is completed.
Reassignment of partition order-1 is completed.
...

十、kafka-configs.sh

bash

# 看 broker 动态配置
$K/kafka-configs.sh --bootstrap-server $B --entity-type brokers --entity-name 1 --describe

# 改（运行时生效，重启后失效）
$K/kafka-configs.sh --bootstrap-server $B --entity-type brokers --entity-name 1 \
  --alter --add-config log.retention.hours=48

# 改 topic
$K/kafka-configs.sh --bootstrap-server $B --entity-type topics --entity-name order \
  --alter --add-config retention.ms=86400000

# 改客户端配额（限速某个客户端）
$K/kafka-configs.sh --bootstrap-server $B --entity-type clients --entity-name order-service \
  --alter --add-config 'producer_byte_rate=10485760,consumer_byte_rate=20971520'

broker 动态配置的重启陷阱

--entity-type brokers 改的是 ZK/KRaft 里的动态配置，重启不会丢，但不会写进 server.properties。下次换机器按文件部署时，这些动态配置就没了。改完要在配置文件里同步一份。

十一、kafka-delete-records.sh

按 offset 截断数据，用于"某个分区写了脏数据要删掉"。

bash

cat > /tmp/delete.json <<'EOF'
{"partitions":[{"topic":"order","partition":0,"offset":1200}],"version":1}
EOF

$K/kafka-delete-records.sh --bootstrap-server $B --offset-json-file /tmp/delete.json

text

Executing records delete operation
Records delete operation completed:
partition: order-0	 low_watermark: 1200

意思是分区 0 的 offset 1200 之前全部删除，之后从 1200 开始消费。消费者如果还停在更早的 offset，会收到 OffsetOutOfRangeException 然后按 auto.offset.reset 处理。

这是不可逆的物理删除

没有回收站。执行前确认：

--offset 写的是要保留的起点，不是要删的条数。
有关联的消费组，提前重置位移。
有备份或能从上游重放。

十二、组合起来看集群

bash

B=10.0.0.31:9092
K=/opt/kafka/bin

echo "== brokers =="
$K/kafka-broker-api-versions.sh --bootstrap-server $B | grep -c 'id:'

echo "== under replicated =="
$K/kafka-topics.sh --bootstrap-server $B --describe --under-replicated-partitions

echo "== topics =="
$K/kafka-topics.sh --bootstrap-server $B --list --exclude-internal | wc -l

echo "== top lag =="
$K/kafka-consumer-groups.sh --bootstrap-server $B --all-groups --describe 2>/dev/null \
  | awk 'NR>1 && $6 ~ /^[0-9]+$/ {s[$1]+=$6} END {for (g in s) print s[g], g}' | sort -rn | head -5

text

== brokers ==
3
== under replicated ==
== topics ==
42
== top lag ==
182400 order-group

四行输出，集群状态一眼看完。under replicated 是空的，brokers 是 3，就基本没事。

十三、踩过的坑

现象	根因	处置
`--zookeeper` 报 deprecation	3.x 已改	全部换 `--bootstrap-server`
`--from-beginning` 无效	group 已有提交位移	换新 group 名或显式 `--offset 0`
`reset-offsets` 执行后没变化	还有消费者在线	先停消费服务
`describe --group` 全 `-`	没有活跃消费者	查消费服务进程
`LAG` 为负	`read_committed` 下用 LSO 计算	看 `LOG-END-OFFSET` 时注意隔离级别
`delete` 后 topic 还在列表	删除是异步的，或有活跃生产者	等，或查 `delete.topic.enable`
重分配跑到一半想取消	只能等，或反向再分配	加 `--throttle` 从一开始控制影响
`dump-log` 输出太多	没加 `head`	用 `--max-message-size` 或直接 `head`

命令行工具：生产、消费、查 lag ​

场景 ​

一、kafka-topics.sh ​

二、kafka-console-producer.sh ​

三、kafka-console-consumer.sh ​

四、kafka-consumer-groups.sh ​

重置位移 ​

删除 group ​

五、kafka-get-offsets.sh ​

六、kafka-log-dirs.sh ​

七、kafka-dump-log.sh ​

八、kafka-leader-election.sh ​

九、kafka-reassign-partitions.sh ​

1. 准备 topic 列表 ​

2. 生成方案 ​

3. 执行与校验 ​

十、kafka-configs.sh ​

十一、kafka-delete-records.sh ​

十二、组合起来看集群 ​

十三、踩过的坑 ​

命令行工具：生产、消费、查 lag

场景

一、kafka-topics.sh

二、kafka-console-producer.sh

三、kafka-console-consumer.sh

四、kafka-consumer-groups.sh

重置位移

删除 group

五、kafka-get-offsets.sh

六、kafka-log-dirs.sh

七、kafka-dump-log.sh

八、kafka-leader-election.sh

九、kafka-reassign-partitions.sh

1. 准备 topic 列表

2. 生成方案

3. 执行与校验

十、kafka-configs.sh

十一、kafka-delete-records.sh

十二、组合起来看集群

十三、踩过的坑