监控与日常运维

# 临时查看
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group inventory-service

# broker 启动参数加 (jmx_prometheus_javaagent)
export KAFKA_OPTS="-javaagent:/opt/jmx_prometheus_javaagent.jar=7071:/opt/kafka-jmx.yml"

# kafka-jmx.yml 片段: 白名单方式只采关键指标
lowercaseOutputName: true
rules:
  - pattern: kafka.server<type=ReplicaManager, name=(UnderReplicatedPartitions|UnderMinIsrPartitionCount)><>Value
    name: kafka_server_replicamanager_$1
  - pattern: kafka.controller<type=KafkaController, name=(ActiveControllerCount|OfflinePartitionsCount)><>Value
    name: kafka_controller_$1
  - pattern: kafka.server<type=BrokerTopicMetrics, name=(BytesInPerSec|BytesOutPerSec)><>OneMinuteRate
    name: kafka_server_brokertopicmetrics_$1

# prometheus.yml
scrape_configs:
  - job_name: kafka-broker
    static_configs: [{ targets: ["kafka1:7071", "kafka2:7071", "kafka3:7071"] }]
  - job_name: kafka-exporter
    static_configs: [{ targets: ["kafka-exporter:9308"] }]

groups:
  - name: kafka
    rules:
      - alert: KafkaUnderReplicated
        expr: sum(kafka_server_replicamanager_underreplicatedpartitions) > 0
        for: 5m
      - alert: KafkaNoActiveController
        expr: sum(kafka_controller_activecontrollercount) != 1
        for: 1m
      - alert: ConsumerLagGrowing
        expr: sum by (consumergroup, topic)
              (delta(kafka_consumergroup_lag[10m])) > 0
        for: 30m

# 1. 生成方案: 把 topics.json 里的 topic 均衡到 broker 1,2,3,4
cat > topics.json <<'EOF'
{"topics": [{"topic": "orders"}], "version": 1}
EOF
bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --topics-to-move-json-file topics.json \
  --broker-list "1,2,3,4" --generate
# 输出两段 JSON: 当前分布 (留作回滚) 与建议方案 (存为 plan.json)
 
# 2. 执行, 限速 50MB/s 防止同步流量打爆网卡与磁盘
bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --reassignment-json-file plan.json --throttle 52428800 --execute
 
# 3. 验证 (完成后会自动移除限速; 没完成前反复跑)
bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --reassignment-json-file plan.json --verify

1. 新 broker 配好同一集群 ID 与 controller.quorum.voters, 启动入集群
2. kafka-reassign-partitions 把部分分区迁到新节点 (限速!)
3. verify 完成, 观察 UnderReplicatedPartitions 归零

1. 生成不含目标 broker 的重分配方案并执行
2. 确认该 broker 上已无任何分区副本
3. 优雅停止 (见下), 再从监控与配置中移除

# 每台机器重复以下步骤
# 1. 健康前提: 无 under-replicated 分区
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --under-replicated-partitions   # 应无输出
 
# 2. 优雅停止 (发 SIGTERM, 触发 controlled shutdown: leader 主动转移)
bin/kafka-server-stop.sh
 
# 3. 升级/改配置后启动, 等它重新加入所有 ISR
# 4. 确认 UnderReplicatedPartitions 归零后, 再做下一台