1. 变更流程

1.1. Leader partition 迁移

1.1.1. 环境及脚本准备

ssh root@172.16.xx.xx
cd /root/zhaobo/offline-kafka

# 新建生成迁移策略脚本
# 初始化运行环境，新建以下目录
mkdir under_switch_leader_partitions

在 under_switch_leader_partitions 新建partition_list、reassign_partitions.sh、switch_leader.sh三个文件。

cd under_switch_leader_partitions
# partition_list 为空文件，用来存放待迁移 Leader partition 列表。
touch partition_list
touch reassign_partitions.sh
touch switch_leader.sh

reassign_partitions.sh 脚本内容如下：

#/bin/bash# e.g.
# sh reassign_partitions.sh 'zookeeper.c.t.xxxxapp-inc.cn:2181' 'partition_list'zookeeper=$1
partition_list=$2for partition in `cat $partition_list`;
doecho "********** Start reassign ${partition}**********"sh /opt/app/kafka/bin/kafka-reassign-partitions.sh --zookeeper $1 --reassignment-json-file ${partition}_partition2reassign.json --execute —throttle 50000000while :dosleep 1status=`sh /opt/app/kafka/bin/kafka-reassign-partitions.sh --zookeeper $1 --reassignment-json-file ${partition}_partition2reassign.json --verify|grep 'Reassignment of partition'`expect_status="Reassignment of partition "$partition" completed successfully"echo "********** Checking status of ${partition} **********"if [ "$status" == "$expect_status" ];thenecho "********** ${partition} meet the expectation **********"breakfidoneecho '********** Sleep 10s **********'sleep 10
done

switch_leader.sh 脚本内容如下：

#!/bin/bash# e.g.
# sh switch_leader.sh 'zookeeper.c.t.xxxxapp-inc.cn:2181' 'partition_list'zookeeper=$1
partition_list=$2for partition in `cat $partition_list`;
doecho "********** ${partition} start leader election **********"sh /opt/app/kafka/bin/kafka-preferred-replica-election.sh --zookeeper $1 --path-to-json-file ${partition}_leader2election.jsonecho '********** Sleep 10s **********'sleep 10
done

1.1.2. 生成迁移策略

修改 xxxx_repair_kafka_disk_offline-kafka.py 脚本参数，broken_broker_id、health_broker_id和bootstrap_servers。

broken_broker_id：异常 Broker ID。
health_broker_id：正常 Broker ID。
bootstrap_servers：可访问的目的集群 bootstrap servers 连接地址。

执行 xxxx_repair_kafka_disk_offline-kafka.py 生成 Leader partition 迁移策略，策略生成到 under_switch_leader_partitions 目录中。

python3 xxxx_repair_kafka_disk_offline-kafka.py

xxxx_repair_kafka_disk_offline-kafka.py 脚本内容如下：

#!/usr/bin/env python
# coding=utf-8import json
import random
import osfrom kafka import KafkaAdminClientdef get_under_reassign_partition():# 输入坏节点 Broker IDbroken_broker_id = [3, 26, 28, 30, 103, 111, 114]# 必须保证副本充足health_broker_id = [1, 2, 5, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 27, 29, 31, 32, 34, 35, 100, 101, 102, 104, 105, 106, 107, 108, 109, 110, 112, 113, 115, 116, 117, 118, 119]# 输入待检测集群 Bootstrap Serveradmin_client = KafkaAdminClient(bootstrap_servers='172.16.xx.xx:9092')topic_list = admin_client.list_topics()#topic_list = ['frequent-control-test-common-changelog', 'frequent-control-test2-common-changelog']topic_infos = admin_client.describe_topics(topics=topic_list)for topic_info in topic_infos:topic_name = topic_info['topic']for partition_info in topic_info['partitions']:partition_id = partition_info['partition']leader_id = partition_info['leader']replicas = partition_info['replicas']isr = partition_info['isr']if leader_id in broken_broker_id:print('Leader 在 Broker ' + str(broken_broker_id) + '坏节点的 Partition 信息：',topic_name + '-' + str(partition_id), replicas)if len(replicas) >= 2:new_replicas = [broker for broker in reversed(replicas)]if new_replicas[0] in broken_broker_id:under_reassign_broker_list = [broker_id for broker_id in health_broker_id ifbroker_id not in new_replicas]random_broker_id = random.sample(under_reassign_broker_list, k=1)for broker_id in random_broker_id:new_replicas[0] = broker_idpartition2reassign_json = {"version": 1, "partitions": [{"topic": topic_name, "partition": partition_id, "replicas": new_replicas}]}leader2election_json = {"version": 1,"partitions": [{"topic": topic_name, "partition": partition_id}]}print(json.dumps(partition2reassign_json))print(json.dumps(leader2election_json))reassign_file_name = topic_name + '-' + str(partition_id) + '_partition2reassign.json'leader_election_file_name = topic_name + '-' + str(partition_id) + '_leader2election.json'if not os.path.exists('under_switch_leader_partitions' + '/' + reassign_file_name):with open('under_switch_leader_partitions' + '/' + reassign_file_name, 'wt') as f:f.write(json.dumps(partition2reassign_json))else:print('File already exists')if not os.path.exists('under_switch_leader_partitions' + '/' + leader_election_file_name):with open('under_switch_leader_partitions' + '/' + leader_election_file_name, 'wt') as f:f.write(json.dumps(leader2election_json))else:print('File already exists')if __name__ == '__main__':get_under_reassign_partition()

以上脚本生成两类文件，分别为：{TOPIC_NAME}-{PARTITION_ID}_partition2reassign.json和{TOPIC_NAME}-{PARTITION_ID}_leader2election.json

{TOPIC_NAME}-{PARTITION_ID}_partition2reassign.json 为分区 reassign 策略文件。
{TOPIC_NAME}-{PARTITION_ID}_leader2election.json 为执行 Partition Leader 切换文件。

将 {TOPIC_NAME}-{PARTITION_ID} 写入 partition_list 文件。

1.1.3. 执行迁移策略

执行重新分配 Leader Partition，reassign_partitions.sh 脚本。

sh reassign_partitions.sh 'zookeeper.c.t.xxxxapp-inc.cn:2181' 'partition_list'

执行 Leader 切换，switch_leader.sh 脚本。

sh switch_leader.sh 'zookeeper.c.t.xxxxapp-inc.cn:2181' 'partition_list'

1.2. 注释 /etc/fstab

vim /etc/fstab

# 按照阿里云要求，将故障磁盘注释，举例如下：#
# /etc/fstab
# Created by anaconda on Wed Dec 25 03:18:13 2019
#
# Accessible filesystems, by reference, are maintained under '/dev/disk'
# See man pages fstab(5), findfs(8), mount(8) and/or blkid(8) for more info
#
UUID=8770013a-4455-4a77-b023-04d04fa388c8 /                       ext4    defaults        1 1#/dev/vdb /opt/data/kafka1     ext4 defaults,noatime 0 0
/dev/vdc /opt/data/kafka2     ext4 defaults,noatime 0 0
/dev/vdd /opt/data/kafka3     ext4 defaults,noatime 0 0
/dev/vde /opt/data/kafka4     ext4 defaults,noatime 0 0
/dev/vdf /opt/data/kafka5     ext4 defaults,noatime 0 0
/dev/vdg /opt/data/kafka6     ext4 defaults,noatime 0 0
#/dev/vdh /opt/data/kafka7     ext4 defaults,noatime 0 0
/dev/vdi /opt/data/kafka8     ext4 defaults,noatime 0 0

1.3. server.properties 删除故障磁盘

# log.dirs 删除故障磁盘，举例如下：
log.dirs=/opt/data/kafka2/log,/opt/data/kafka3/log,/opt/data/kafka4/log,/opt/data/kafka5/log,/opt/data/kafka6/log,/opt/data/kafka8/log

1.4. 重启 Broker

# 停止 Broker
/opt/app/kafka/bin/kafka-server-stop.sh

# 启动 Broker
/opt/app/kafka/bin/kafka-server-start.sh -daemon /opt/app/kafka/config/server.properties

Broker 重启后，数据重新加载，根据数据量不同，磁盘有IO升高的情况。

1.5. umount 故障磁盘

# umount 磁盘
umount -v /opt/data/kafka

1.6. 阿里云控制台报修

打开故障磁盘控制台，根据提示报修磁盘，修复时间不确定，建议添加报警提示。

1.7. 已修复磁盘挂载

打开故障磁盘控制台，根据提示修复磁盘。参考阿里云文档：https://help.aliyun.com/document_detail/34377.html

1.8. server.properties 添加修复故障磁盘

# log.dirs 添加修复的故障磁盘，举例如下：
log.dirs=/opt/data/kafka2/log,/opt/data/kafka3/log,/opt/data/kafka4/log,/opt/data/kafka5/log,/opt/data/kafka6/log,/opt/data/kafka8/log

1.9. 重启 Broker

# 停止 Broker
/opt/app/kafka/bin/kafka-server-stop.sh# 启动 Broker
/opt/app/kafka/bin/kafka-server-start.sh -daemon /opt/app/kafka/config/server.properties

Broker 重启后，数据重新加载，根据数据量不同，磁盘有IO升高的情况。

1.10. 均衡 Kafka 集群压力

ssh root@172.16.xx.xx
cd /root/zhaobo/offline-kafka

新建待迁移 Topic 列表 JSON 模板

{"topics":[{"topic":"topic_template"}],"version":1
}

新建 Topic 列表topic_not_repair.list

# e.g.
# cat topic_not_repair.list
zb_test1

生成可执行 Topic 分区 JSON 文件

# e.g.
# sh create_exec_json_new.sh 'topic_not_repair.list' '172.16.xx.xx:2181/log-kafka' '1,2,3,4,5,6,7,8,9,10,11,12,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65'#!/bin/bashfor i in `cat $1`;
docat topic4move.json_template | sed "s/topic_template/$i/g" >> $i'_topic4move.json'sleep 1sh /opt/app/kafka/bin/kafka-reassign-partitions.sh --zookeeper $2 --topics-to-move-json-file $i'_topic4move.json' --broker-list $3 --generate|awk 'END {print}' >> $i'_topic2exec.json';done

执行脚本，生成可执行重分配 JSON 文件

sh create_exec_json_new.sh 'topic_not_repair.list' '172.16.xx.xx:2181/kafka' '1,3,6,8,11,12,13,14,18,19,20,21,23,24,25,26,27,28,29,30,31,32,33,34,35,100,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,116,117,118,119'

执行重新分配，记录打印内容，里边包含回滚配置。

/opt/app/kafka/bin/kafka-reassign-partitions.sh --zookeeper 172.16.xx.xx:2181/kafka --reassignment-json-file ck-test1_topic2exec.json --execute

重新分配日志包含回滚配置
数据量大的情况下需要修改数据保留时间，减轻磁盘压力：

/opt/app/kafka/bin/kafka-topics.sh --zookeeper 172.16.xx.xx:2181/kafka -topic ck-test1 --alter --config retention.ms=43200000`

2. 校验流程

ECS 控制台坏盘事件消失。
Broker 所在 ECS 的 /var/logs/message 无坏盘日志。
Broker 日志无坏盘报错。

3. 回滚流程

修复流程，无需回滚。

Kafka （阿里云 ECS）磁盘类故障修复操作文档相关推荐

在阿里云进行磁盘扩容操作后，ECS实际使用磁盘并没有扩容？阿里云ECS磁盘扩容操作
阿里云ECS磁盘扩容在阿里云对ECS进行磁盘扩容. 在线磁盘扩容查看实例的云盘情况 fdisk -lu 确认已有分区的文件系统类型 df -Th 检查到我的系统盘在 /dev/vda1 安装工具 ...
块存储性能--阿里云ECS磁盘
块存储性能更新时间:2020-06-15 17:52:20 https://help.aliyun.com/document_detail/25382.html?spm=5176.ecsbuyv3. ...
阿里云ECS服务器提示漏洞修复问题自己手动修复
修复步骤参考: 一.root登陆终端二.yum check-update 查看可升级的系统软件三.yum upgrade 升级所有可升级的系统软件四.y 确认升级五.直至显示Complete! ...
阿里云ecs Windows条带化方法及优缺点
一. 背景系统对磁盘速度要求较高,但测试时发现阿里云ecs ssd磁盘只有300M/s.查询文档发现阿里云单块磁盘速度有上限,高级essd能达到速度要求但价格每个月增加约1/3,测试并记录下Wind ...
七牛云存储挂载到阿里云ECS上的目录下，达到扩容磁盘的效果
七牛云存储挂载到阿里云ECS上,达到扩容磁盘的效果 1. 安装环境 yum install automake fuse fuse-devel gcc-c++ git libcurl-devel lib ...
阿里云ECS云服务器选型教程
本文主要大概分为4个部分,第一部分介绍云服务器ECS的基本概念:第二部分对ECS的实例规格族进行一个详细的解读:第三部分给大家实战讲解如何去做ECS实例的选型:最后一部分简单介绍一下如何去省钱省力的来 ...
怎么把springboot部署到阿里云_听说“一分钟就能部署阿里云ECS集群”？
导读 "企业上云"是当下大势所趋,"提效.降成本" 也是企业上云.迁云和管理云必须关注的问题.今天我们将从一个集群部署的场景出发,介绍阿里云如何借助生态工具Te ...
阿里云ECS云服务器快照概念以及使用（六）
阿里云ECS云服务器快照概念以及使用文章目录阿里云ECS云服务器快照概念以及使用 1.快照的概念 1.1.什么是快照 1.2.云盘增量快照 2.为云盘手动创建快照 2.1.创建快照 2.3.模拟E ...
阿里云ECS问题大全【转自阿里云社区】
云服务器ECS是阿里云的明星产品,上云基础,阿里云ECS不仅有面向企业场景的计算实例.存储及网络组件,使应用运行稳定.数据存储可靠.有着丰富的安全防护及监控措施,而且可以使用多种管理方式,赋予完全管理 ...

Kafka （阿里云 ECS）磁盘类故障修复操作文档