Kafka运维宝典 （二）- kafka 查看kafka的运行状态、broker.id不一致导致启动失败问题、topic消息积压量告警监控脚本

Kafka运维宝典 （二）

一、kafka broker.id冲突问题

Kafka 中的 broker.id 是每个 Kafka Broker 的唯一标识符，用于区分集群中的各个 Broker。当 Kafka 集群启动时，每个 Broker 必须有一个唯一的 broker.id，用于确保集群中各个 Broker 的正确识别与协作。如果 broker.id 冲突，Kafka 将无法正确识别集群中的 Broker，可能导致集群的异常行为或启动失败。

1. broker.id 冲突的影响

当两个或多个 Kafka Broker 配置了相同的 broker.id 时，会发生以下问题：

• 集群启动失败：Kafka 在启动时会检查 broker.id 是否唯一。如果发现冲突，Broker 启动时会抛出错误，导致集群无法正常运行。
• 数据丢失：如果两个 Broker 使用相同的 broker.id，可能会导致分区副本的写入错误，从而引发数据丢失或不一致。
• 网络连接问题：由于 broker.id 冲突，集群中的其他 Broker 可能无法正确与重复 broker.id 的 Broker 进行通信，从而影响集群的健康和通信。

2. 如何发现 broker.id 冲突

当 Kafka Broker 启动时，如果存在 broker.id 冲突，日志中通常会有明确的错误提示。你可以通过查看 Kafka 的 server.log 文件来查找这些错误。

日志输出（broker.id 冲突）：

[2025-01-20 08:00:00,123] ERROR Broker id 1 already exists (kafka.server.KafkaServer)
[2025-01-20 08:00:00,124] ERROR Error processing KafkaServer (kafka.server.KafkaServer)

该错误表示 broker.id 为 1 的 Broker 已经存在，另一个 Broker 尝试使用相同的 broker.id 启动，从而导致冲突。

3. 解决 broker.id 冲突的方法

解决 broker.id 冲突需要确保集群中每个 Broker 配置的 broker.id 唯一。以下是几个常见的解决方法：

a. 检查和修复配置文件
每个 Kafka Broker 的 broker.id 配置在其配置文件 server.properties 中。确保每个 Broker 配置文件中的 broker.id 唯一。

示例：
假设你有两个 Broker，Broker 1 和 Broker 2：

• Broker 1 的 server.properties：

  broker.id=1
  log.dirs=/var/lib/kafka-logs
  zookeeper.connect=localhost:2181
  

• Broker 2 的 server.properties：

  broker.id=2
  log.dirs=/var/lib/kafka-logs
  zookeeper.connect=localhost:2181
  

检查每个 Broker 的 server.properties 文件，确保 broker.id 唯一。

b. 查看启动的 Kafka 实例
如果 Kafka Broker 是动态启动的，特别是在容器化部署环境中，可能会出现多个 Broker 启动时配置相同的 broker.id。此时可以通过以下方式避免冲突：

• 确保每个 Broker 启动时，broker.id 是唯一的，可以通过环境变量或配置管理工具（如 Kubernetes 配置）来动态设置 broker.id。
• 使用自动化工具（如 Ansible、Chef 或 Puppet）来管理 Kafka 配置，确保每个 Broker 的 broker.id 是唯一的。

c. 使用不同的端口和 IP
如果 Kafka Broker 在同一台机器上启动，可能会由于重复的 broker.id 配置而发生冲突。在这种情况下，可以通过使用不同的端口来启动每个 Broker，并确保每个 Broker 的 broker.id 唯一。

假设有两个 Broker 启动在同一台机器上：

• Broker 1 启动命令：

  bin/kafka-server-start.sh /path/to/config/server-1.properties
  

• Broker 2 启动命令：

  bin/kafka-server-start.sh /path/to/config/server-2.properties
  

在 server-1.properties 和 server-2.properties 中，确保 broker.id 唯一。

d. 在多节点部署中使用唯一的 broker.id
在多节点部署环境中，每个节点的 broker.id 必须唯一。使用 zookeeper 时，Kafka 会基于 broker.id 进行节点识别和通信。确保每个节点的 broker.id 唯一并且符合集群中节点的配置。

4. broker.id 配置管理

对于大型 Kafka 集群，手动管理 broker.id 可能会变得困难。在这种情况下，可以使用集中式配置管理工具来自动分配 broker.id，并避免冲突。

• Kubernetes 配置管理：如果 Kafka 集群部署在 Kubernetes 上，可以通过 StatefulSets 来保证每个 Pod 的 broker.id 唯一性。Kubernetes 会根据 Pod 的名称分配唯一的 broker.id。

• 配置管理工具（如 Ansible）：你可以使用 Ansible、Chef 等工具自动化配置文件的管理，确保每个 Kafka Broker 配置文件中的 broker.id 唯一。

5. 集群启动后确认 broker.id 唯一性

集群启动后，可以通过以下命令确认每个 Broker 的 broker.id 是否唯一，并检查集群状态。

a. 查看集群信息
使用 kafka-broker-api-versions.sh 查看集群中的所有 Broker 的版本信息。

bin/kafka-broker-api-versions.sh --bootstrap-server <broker_host>:<port> --describe

示例输出：

Broker ID    Host:port        Max version   Min version
1            kafka-broker-1:9092   15            0
2            kafka-broker-2:9093   15            0

b. 查看 Broker 日志
查看每个 Broker 的启动日志，确保没有 broker.id 冲突的错误。

tail -f /path/to/kafka/logs/server.log

6. broker.id 冲突的其他常见场景

除了手动配置错误或多个 Broker 启动在同一台机器上时，broker.id 冲突还可能发生在以下场景：

• 容器化部署：如果使用 Docker 或 Kubernetes 部署 Kafka，可能会因为容器重启、Pod 重启等原因，导致 Kafka 实例使用相同的 broker.id。确保每个容器或 Pod 都有唯一的 broker.id。
• 集群扩容时的配置遗漏：在扩展 Kafka 集群时，若新加入的 Broker 配置了与现有 Broker 相同的 broker.id，也会导致冲突。
• 配置文件备份恢复问题：有时候，错误的备份或恢复操作可能会导致不同机器上的 Kafka Broker 使用相同的 broker.id。

二、server.properties的broker.id和meta.properties中的broker.id不一致导致启动失败问题

在 Kafka 集群中，server.properties 文件中的 broker.id 和 meta.properties 文件中的 broker.id 是两个重要的配置项，它们都与 Kafka Broker 的标识密切相关。它们之间必须保持一致，否则会导致 Kafka 启动失败、集群元数据混乱，甚至影响数据的存储和读取。

1. 概述

• server.properties 中的 broker.id：这是 Kafka Broker 的唯一标识符，通常在每个 Broker 启动时由 server.properties 文件中的 broker.id 配置指定。
• meta.properties 中的 broker.id：meta.properties 文件存储了 Kafka Broker 的元数据信息，包括 broker.id 和 cluster.id 等。该文件是 Kafka Broker 启动时从磁盘读取的元数据文件，用于管理集群中的 Broker 信息。

2. broker.id 配置不一致的情况

如果 server.properties 中的 broker.id 和 meta.properties 中的 broker.id 不一致，Kafka 将无法正常启动或工作。以下是具体的情况和例子：

a. 启动时的检查

• server.properties 中的 broker.id：这是 Kafka 启动时从配置文件中读取的 broker.id。Kafka 在启动时会根据 server.properties 中的 broker.id 来配置当前 Broker 的标识。
• meta.properties 中的 broker.id：这是 Kafka 启动时从磁盘中读取的元数据文件。如果 meta.properties 中记录的 broker.id 与 server.properties 中的 broker.id 不一致，Kafka 会检测到这个不匹配并抛出错误，导致启动失败。

3. 实际例子

假设你有两个 Kafka Broker，Broker 1 和 Broker 2，它们的配置文件如下所示：

Broker 1 的配置 (server-1.properties)：

broker.id=1
listeners=PLAINTEXT://192.168.1.1:9092
advertised.listeners=PLAINTEXT://192.168.1.1:9092
zookeeper.connect=192.168.1.10:2181

Broker 2 的配置 (server-2.properties)：

broker.id=2
listeners=PLAINTEXT://192.168.1.2:9092
advertised.listeners=PLAINTEXT://192.168.1.2:9092
zookeeper.connect=192.168.1.10:2181

在这个配置中，broker.id=1 和 broker.id=2 是唯一且正确的。然而，假设我们在运行时发生了某些问题，例如：

Broker 1 启动时的 meta.properties 文件
假设 Broker 1 启动后，在其数据目录下生成了一个 meta.properties 文件，内容如下：

broker.id=2
cluster.id=some-cluster-id

4. 发生的问题

此时，我们就会遇到 server.properties 中的 broker.id 和 meta.properties 中的 broker.id 不一致的问题。具体来说：

1. 启动时冲突：当 Broker 1 启动时，Kafka 会首先从 server.properties 中读取 broker.id=1，然后检查 meta.properties 中存储的 broker.id=2。由于两者不一致，Kafka 会检测到配置冲突，抛出启动错误，无法正常启动。

2. 错误信息：在 Kafka 的启动日志中，你可能会看到类似以下的错误信息：

   [ERROR] Broker id 1 does not match the metadata file id 2 (kafka.server.KafkaServer)
   

3. 元数据管理失败：由于 broker.id 不一致，Kafka 会无法将 Broker 1 正确地与集群中的其他 Broker 同步元数据。这样就无法正常进行 Leader 选举和副本同步，导致集群的元数据不一致。

5. 导致的后果

• Kafka 无法启动：server.properties 和 meta.properties 中的 broker.id 不一致会导致 Kafka Broker 启动失败。
• 元数据不同步：即使集群中的其他 Broker 可以正常启动，但由于 broker.id 不一致，集群的元数据无法正确同步，Kafka 可能无法正确管理分区、副本等。
• Leader 选举失败：由于集群中的某些 Broker 的 broker.id 不一致，可能导致 Leader 选举失败，进而导致消费者无法读取数据或者生产者无法写入数据。

6. 如何解决 broker.id 不一致问题

a. 检查并修正 server.properties 中的配置
确保 server.properties 中的 broker.id 配置与其他 Broker 的配置一致且唯一。例如，确保 broker.id=1 和 broker.id=2 是唯一的。

broker.id=1
listeners=PLAINTEXT://192.168.1.1:9092
advertised.listeners=PLAINTEXT://192.168.1.1:9092
zookeeper.connect=192.168.1.10:2181

b. 检查 meta.properties 文件中的 broker.id
查看 meta.properties 文件中的 broker.id，确保它与 server.properties 中的配置一致。如果不一致，可以手动修改 meta.properties 文件中的 broker.id。

broker.id=1
cluster.id=some-cluster-id

c. 重新初始化 meta.properties
如果无法直接修改 meta.properties，可以通过删除 meta.properties 文件并重新启动 Kafka 来重建该文件。在 Kafka 启动时，系统会根据新的 server.properties 文件生成新的 meta.properties。

删除 meta.properties 文件：

rm /path/to/kafka/data/meta.properties

然后，重新启动 Kafka Broker：

bin/kafka-server-start.sh config/server-1.properties

d. 确保 ZooKeeper 中的 Broker 信息一致
在使用 ZooKeeper 作为 Kafka 集群的元数据存储时，确保 ZooKeeper 中记录的 broker.id 是一致的。你可以使用 ZooKeeper CLI 检查注册的 Broker 信息：

zkCli.sh -server <zookeeper_host>:2181
ls /brokers/ids

输出：

[1, 2]

确保没有重复的 broker.id，如果有，可以通过删除重复的节点来解决。

e. 重启 Kafka 集群
修复 server.properties 和 meta.properties 后，重启 Kafka 集群中的所有 Broker 来确保配置生效。

bin/kafka-server-start.sh config/server-1.properties
bin/kafka-server-start.sh config/server-2.properties

7. 防止未来的 broker.id 不一致问题

• 配置管理：使用配置管理工具（如 Ansible、Chef、Puppet）来确保 server.properties 和 meta.properties 中的 broker.id 一致。
• 环境变量：在容器化或虚拟化环境中，可以通过环境变量动态分配 broker.id，并确保一致性。
• 自动化监控和告警：设置 Kafka 集群的监控和告警系统，在 broker.id 配置异常时提前发现并处理。

三、Kafka topic消息积压量告警监控脚本

Kafka 中的 消息积压量（Consumer Lag） 是指消费者组的消费进度与 Kafka 集群中消息的生产进度之间的差距。监控消息积压量对于 Kafka 集群的健康至关重要，因为过高的积压量可能导致系统性能下降、延迟增大，甚至引发数据丢失。

在这个脚本中，我们将创建一个监控每个消费者组的 滞后量（Lag）的脚本，并在滞后量超过预设阈值时发送告警通知。

1. 监控目标

• 监控目标：检测每个消费者组的滞后量（Lag）。
• 滞后量计算：滞后量（Lag）是消费者组当前消费进度与最新生产者消息写入进度之间的差距。具体地，滞后量 = Log End Offset - Consumer Offset。
• 告警阈值：当消费者组的滞后量超过预设的阈值时，触发告警。

2. 监控原理

Kafka 的消费者滞后量是通过以下两者计算的：

• Log End Offset (LEO)：表示每个分区的最新消息偏移量。
• Consumer Offset：表示消费者组已经消费的消息的偏移量。

对于每个消费者组，监控脚本会定期检查每个分区的消费者滞后量，并判断是否超过设定的阈值。如果超过阈值，脚本会发送告警邮件。

3. 监控脚本

以下是一个用于监控每个 Kafka 消费者组滞后量的 Bash 脚本示例。该脚本会：

1. 获取每个分区的消费偏移量和日志末尾偏移量。
2. 计算每个消费者组的滞后量（Lag）。
3. 如果滞后量超过阈值，发送告警。

#!/bin/bash

# Kafka 集群信息
KAFKA_BROKER="localhost:9092"
LAG_THRESHOLD=1000  # 设置滞后量阈值（单位：消息数），如：1000
ALERT_EMAIL="your_alert_email@example.com"  # 告警邮件收件人

# 获取所有消费者组列表
CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server $KAFKA_BROKER --list)

# 遍历每个消费者组并检查滞后量
for GROUP in $CONSUMER_GROUPS
do
    echo "Checking lag for consumer group: $GROUP"

    # 获取该消费者组所有分区的滞后量
    LAG_SUM=0
    PARTITIONS=$(kafka-consumer-groups.sh --bootstrap-server $KAFKA_BROKER --describe --group $GROUP | awk '{print $1, $2}')
    
    while read -r line; do
        # 解析每个分区的日志末尾偏移量 (LEO) 和消费者偏移量
        PARTITION=$(echo $line | awk '{print $1}')
        LEO=$(echo $line | awk '{print $4}')
        CONSUMER_OFFSET=$(echo $line | awk '{print $5}')
        
        # 计算该分区的滞后量
        LAG=$(($LEO - $CONSUMER_OFFSET))
        LAG_SUM=$(($LAG_SUM + $LAG))

        echo "Partition: $PARTITION, Log End Offset: $LEO, Consumer Offset: $CONSUMER_OFFSET, Lag: $LAG"
    done <<< "$PARTITIONS"

    echo "Total Consumer Lag for group $GROUP: $LAG_SUM"

    # 判断滞后量是否超过阈值
    if [ "$LAG_SUM" -gt "$LAG_THRESHOLD" ]; then
        echo "WARNING: Total Consumer Lag for group $GROUP is $LAG_SUM, which exceeds the threshold $LAG_THRESHOLD" | mail -s "Kafka Consumer Lag Alert" $ALERT_EMAIL
        echo "Alert sent for $GROUP!"
    else
        echo "Consumer Lag for group $GROUP is within acceptable limits."
    fi
done

1. KAFKA_BROKER：Kafka 集群的 Broker 地址，格式为 localhost:9092。
2. LAG_THRESHOLD：滞后量阈值，如果某个消费者组的滞后量总和超过该值，脚本会触发告警。
3. ALERT_EMAIL：告警邮件的收件人地址，用于接收滞后量超标的通知。
4. CONSUMER_GROUPS：通过 kafka-consumer-groups.sh --list 命令获取所有消费者组的列表。
5. kafka-consumer-groups.sh --describe --group $GROUP：获取指定消费者组的分区信息，包括日志末尾偏移量（LEO）和消费者偏移量。
6. LAG_SUM：该变量存储所有分区的滞后量总和。
7. 滞后量计算：
   • 对于每个分区，脚本通过计算 LEO - Consumer Offset 得到该分区的滞后量。
   • 将所有分区的滞后量相加，得到该消费者组的总滞后量。
8. 告警：
   • 如果消费者组的总滞后量超过阈值，脚本会通过 mail 命令发送告警邮件。

脚本运行时的输出示例如下：

Checking lag for consumer group: your_consumer_group
Partition: 0, Log End Offset: 50000, Consumer Offset: 49000, Lag: 1000
Partition: 1, Log End Offset: 52000, Consumer Offset: 51000, Lag: 1000
Total Consumer Lag for group your_consumer_group: 2000
WARNING: Total Consumer Lag for group your_consumer_group is 2000, which exceeds the threshold 1000
Alert sent for your_consumer_group!

如果滞后量未超过阈值，输出将是：

Consumer Lag for group your_consumer_group is within acceptable limits.

4. 定时任务

为了让脚本定期运行并监控消费者组的滞后量，可以将它添加到 cron 定时任务中。例如，以下是每分钟执行一次该脚本的 cron 配置：

* * * * * /path/to/kafka-lag-monitor.sh