今天kafka測試環境中機器磁碟告警，佔用率超過了80%，原來是某一個topic的partition為1，只往一臺機器上寫資料，造成kafka叢集空間使用不均。
下面主要使用kafka-topics.sh和kafka-reassign-partitions.sh來解決問題。
推薦使用kafka manager來管理kafka叢集。

修改topic的partitions

此命令執行完之後即可再kafka叢集其他機器中找到此topic的目錄

擴容、刪除機器

只要配置zookeeper.connect為要加入的叢集，再啟動Kafka程序，就可以讓新的機器加入到Kafka叢集。但是新的機器只針對新的Topic才會起作用，在之前就已經存在的Topic的分割槽，不會自動的分配到新增加的物理機中。為了使新增加的機器可以分擔系統壓力，必須進行訊息資料遷移。Kafka提供了kafka-reassign-partitions.sh進行資料遷移。

這個指令碼提供3個命令：

• --generate: 根據給予的Topic列表和Broker列表生成遷移計劃。generate並不會真正進行訊息遷移，而是將訊息遷移計劃計算出來，供execute命令使用。
• --execute: 根據給予的訊息遷移計劃進行遷移。
• --verify: 檢查訊息是否已經遷移完成。

示例

topic為test目前在broker id為1,2,3的機器上，現又添加了兩臺機器，broker id為4,5，現在想要將壓力平均分散到這5臺機器上。

手動生成一個json檔案topic.json

呼叫--generate生成遷移計劃，將test擴充到所有機器上

生成類似於下方的結果

Current partition replica assignment表示當前的訊息儲存狀況。Proposed partition reassignment configuration表示遷移後的訊息儲存狀況。
將遷移後的json存入一個檔案reassignment.json，供--execute命令使用。

執行--execute進行擴容。

使用--verify檢視進度

相關命令

1. ./bin/kafka-console-producer.sh --broker-list vlnx111111:9092 --topic test
2. ./bin/kafka-console-consumer.sh --zookeeper vlnx111122:2181 --topic test --from-beginning
3. ./bin/kafka-topics.sh --zookeeper vlnx111122:2181 --list
4. ./bin/kafka-topics.sh --zookeeper vlnx111122:2181 --create --replication-factor 2 --partition 6 --topic test
5. ./bin/kafka-topics.sh --zookeeper vlnx111122:2181 --delete --topic test
6. ./bin/kafka-topics.sh --zookeeper vlnx111122:2181 --describe --topic test

參考