# 配置Kafka数据均衡工具
#### 操作场景
该任务指导管理员根据业务需求，在客户端中执行Kafka均衡工具来均衡Kafka集群的负载，一般用于节点的退服、入服以及负载均衡的场景。
#### 前提条件
- MRS集群管理员已明确业务需求，并准备一个Kafka管理员用户（属于kafkaadmin组，集群未启用Kerberos认证（普通模式）时不涉及）。 创建用户相关操作请参考[创建Kafka用户并绑定角色](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_1032.html#mrs_01_1032__section15172119027)。
  
- 已安装客户端，例如安装目录为"/opt/client"，以下操作的客户端目录只是举例，请根据实际安装目录修改。 下载并安装集群客户端的具体操作，请参考[安装MRS集群客户端](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0089.html)。
  
 
#### 操作步骤
1. 登录MRS集群Manager。 
   登录集群Manager具体操作，请参考[访问MRS集群Manager](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0128.html)。
   
   
2. 获取Kafka节点业务IP及端口。 
   1. 选择"集群 \> 服务 \> Kafka \> 实例"，查看并记录任意一个Broker角色实例的业务IP地址。
   
   2. 选择"配置 \> 全部配置"，并根据集群模式获取端口号：
      - 集群已启用Kerberos认证（安全模式）：搜索"sasl.port"参数，查看并记录端口号，默认为21007。
      
      - 集群未启用Kerberos认证（普通模式）：搜索"port"参数，查看并记录端口号，默认为9092。
       
   
   
   
   
3. 获取ZooKeeper节点业务IP及端口。 
   1. 选择"集群 \> 服务 \> ZooKeeper \> 实例"，查看并记录任意一个ZooKeeper角色实例的业务IP地址。
   
   2. 选择"配置 \> 全部配置"，搜索"clientPort"，查看并记录端口号。 创建LTS版本类型集群时，可以选择"组件端口"为"开源"场景或"定制"场景，选择"开源"时ZooKeeper默认端口为2181，选择"定制"时ZooKeeper默认端口为24002。
      
   
   
   
   
4. 以客户端安装用户，登录安装客户端的节点。
5. 执行以下命令，切换到客户端安装目录，例如安装目录为"/opt/client"，具体以实际替换。 
   ```
   cd /opt/client
   ```
   
   
6. 执行以下命令配置环境变量。 
   ```
   source bigdata_env
   ```
   
   
7. 执行以下命令，进行用户认证。（集群未启用Kerberos认证（普通模式）时跳过此步骤） 
   ```
   kinit 组件业务用户
   ```
   
   
8. 执行以下命令进入"kafka"目录。 
   ```
   cd Kafka/kafka
   ```
   
   
9. 使用"kafka-balancer.sh"进行用户集群均衡，常用命令如下： 
   - 使用--run命令执行集群均衡：
     ```
     ./bin/kafka-balancer.sh --run --zookeeper ZooKeeper的任意一个节点的业务IP:zkPort/kafka --bootstrap-server Kafka集群IP:port --throttle 10000000 --consumer-config config/consumer.properties --show-details
     ```
     - 该命令包含均衡方案的生成和执行两部分。
     
     - --show-details为可选参数，表示是否打印方案明细。
     
     - --throttle表示均衡方案执行时的带宽限制，单位：bytes/sec。
      
   
   - 使用--run命令执行节点退服：
     ```
     ./bin/kafka-balancer.sh --run --zookeeper ZooKeeper的任意一个节点的业务IP:zkPort/kafka --bootstrap-server Kafka集群IP:port --throttle 10000000 --consumer-config config/consumer.properties --remove-brokers <BrokerId列表> --force
     ```
     - 此退服命令会将待退服Broker节点上的数据迁移至其他Broker节点。
     
     - --remove-brokers表示要删除的BrokerId列表，多个间用逗号分隔。
     
     - --force参数为可选参数，表示忽略磁盘使用率告警，强制生成迁移方案。
      
   
   - 查看执行状态：
     ```
     ./bin/kafka-balancer.sh --status --zookeeper ZooKeeper的任意一个节点的业务IP:zkPort/kafka
     ```
     
   
   - 生成均衡方案：
     ```
     ./bin/kafka-balancer.sh --generate --zookeeper ZooKeeper的任意一个节点的业务IP:zkPort/kafka --bootstrap-server Kafka集群IP:port --consumer-config config/consumer.properties
     ```
     该命令仅根据集群当前状态生成迁移方案，并打印到控制台。
     
   
   - 清理中间状态
     ```
     ./bin/kafka-balancer.sh --clean --zookeeper ZooKeeper的任意一个节点的业务IP:zkPort/kafka
     ```
     该命令一般在迁移没有正常执行完成时用来清理ZooKeeper上的中间状态信息。
     
   
   
   
   
 
#### 常见问题
在使用Kafka均衡工具进行Partition迁移的过程中，如果出现集群中Broker故障导致均衡工具的执行进度阻塞，可以根据实际场景，参考对应步骤进行处理：
#### Broker因为磁盘占有率达到100%导致Broker故障
1. 登录FusionInsight Manager，选择"集群 \> 服务 \> Kafka \> 实例"，将运行状态为"正在恢复"的Broker实例停止，并记录实例所在节点的管理IP地址。单击该角色名称，在"实例配置"页面中选择"全部配置"，搜索"broker.id"参数，并记录参数值。 broker.id：表示服务器上Broker的ID值，每个Broker具有唯一的ID值。不支持配置，由系统自动生成。
   
2. 以**root** 用户登录记录的Broker管理IP地址，并执行以下命令，查看磁盘占用率为100%的挂载目录。
   ```
   df -lh
   ```
   例如记录的目录为"${BIGDATA_DATA_HOME}/kafka/data1"，具体以实际为准。
   
3. 执行以下命令，进入磁盘占用率为100%的挂载目录（例如"${BIGDATA_DATA_HOME}/kafka/data1"），并查看该目录下各文件夹的大小。
   ```
   cd ${BIGDATA_DATA_HOME}/kafka/data1
   ```
   ```
   du -sh *
   ```
   查看是否存在除"kafka-logs"目录外的其他文件，并判断是否可以删除或者迁移。
   - 是，删除或者迁移相关数据，然后执行[8]。
   
   - 否，执行[4]。
    
4. 进入"kafka-logs"目录，执行以下命令，选择一个待移动的Partition文件夹，其名称命名规则为"Topic名称-Partition标识"，记录Topic及Partition。
   ```
   du -sh *
   ```
   
5. 修改"kafka-logs"目录下的"recovery-point-offset-checkpoint"和"replication-offset-checkpoint"文件（两个文件做同样的修改）。
   1. 减少文件中第二行的数字（如果移出多个目录，则减少的数字为移出的目录个数）。
   
   2. 删除待移出的Partition所在的行（行结构为"Topic名称 Partition标识 Offset"，删除前先将该行数据保存，后续此内容还要添加到目的目录下的同名文件中）。
    
6. 修改目的数据目录下（例如："${BIGDATA_DATA_HOME}/kafka/data2/kafka-logs"）的"recovery-point-offset-checkpoint"和"replication-offset-checkpoint"文件（两个文件做同样的修改）。
   - 增加文件中第二行的数字（如果移入多个Partition目录，则增加的数字为移入的Partition目录个数）。
   
   - 添加待移入的Partition行到文件末尾（行结构为"Topic名称 Partition标识 Offset"，直接复制[5]中保存的行数据即可）。
    
7. 移动数据，将待移动的Partition文件夹移动到目的目录下，移动完成后执行以下命令修改Partition目录属组。
   ```
   chown omm:wheel -R Partition目录
   ```
   
8. 登录FusionInsight Manager，选择"集群 \> 服务 \> Kafka \> 实例"，启动停止的Broker实例。
9. 等待5至10分钟后查看Broker实例的运行状态是否为"良好"。
   - 是，修复完成后按照"ALM-38001 Kafka磁盘容量不足"告警指导彻底解决磁盘容量不足问题。
   
   - 否，联系运维人员。
    
按照上述步骤将故障Broker进行恢复后，阻塞的均衡任务会继续执行，可使用--status命令来查看任务的执行进度。
#### 问题场景复杂，短时间内无法恢复Broker
由其他原因导致的Broker故障，且问题场景复杂，短时间内无法恢复Broker的情况。
1. 以客户端安装用户，登录安装客户端的节点。
2. 执行以下命令，切换到客户端安装目录，例如安装目录为"/opt/client"，具体以实际替换。
   ```
   cd /opt/client
   ```
   
3. 执行以下命令配置环境变量。
   ```
   source bigdata_env
   ```
   
4. 执行以下命令，进行用户认证。（集群未启用Kerberos认证（普通模式）时跳过此步骤）
   ```
   kinit 组件业务用户
   ```
   
5. 使用以下命令登录ZooKeeper客户端。
   ```
   zkCli.sh -server ZooKeeper集群业务IP:zkPort/kafka
   ```
   
6. 执行**addauth krbgroup**。（普通模式跳过此步骤）
7. 删除"/admin/reassign_partitions"目录和"/controller"目录。
8. 通过以上步骤强行终止迁移，待集群恢复后使用**kafka-reassign-partitions.sh**命令手动将中间过程中导致的多余的副本删除。
 
#### 问题场景单一明确，短时间内可以恢复Broker
由其他原因导致的Broker故障，且问题场景单一明确，短时间内可以恢复Broker的情况。
1. 根据问题根因制定恢复方案，恢复故障Broker。
2. 故障Broker恢复后，阻塞的均衡任务会继续执行，可使用--status命令来查看任务的执行进度。