此工具能做啥?

此工具的作用跟Referred Replica Leader Election工具類似，都是為了平衡叢集的負載。但工具不僅為某一個partition從assigned replica中選舉一個新的leader，還改變partition的assigned replica，回憶一下，follower需要從leader上fetch資料為了保持同步，因此有時僅僅平衡leadershipi的負載是不夠的(還需要考慮follower的負載)。下面是此工具工作的步驟：

1，此工具更新ZK的路徑"/admin/reassign_partitions"，寫入一個topic的partition列表和一個new assigned replica列表(如果設定了json檔案了的話)。

2，controller監聽ZK上的路徑。當資料update時被觸發，controller從ZK讀取更新的topic partition的列表和它們的assigned replicas。

3，對於每一個topic的partition，controller做如下的事情：
3.1. 開始一個新的replicas，由RAR - AR(RAR中存在，AR中不存在的replica)組成(RAR = Reassigned Replicas, AR = original list of Assigned Replicas)。
3.2. 在新的replica跟leader同步之前保持等待狀態。
3.3. 如果leader不在RAR中，從RAR選舉一個新的leader。
3.4 4. 停止老的replicas，由AR - RAR(AR中存在，RAR中不存在的replica)組成。
3.5. 寫入新的AR。
3.6. 從/admin/reassign_partitions刪除partition。

注意此工具只更新zk並退出，controller非同步為partitions reassign replicas。

此工具現在只在0.8分支中提供。

如何使用?

警告：此工具在0.8的beta版本釋出，並有一些bug，在0.8.1版本中是穩定的。

bin/kafka-reassign-partitions.sh
 
Option                                  Description
------                                  -----------
--broker-list <brokerlist>              The list of brokers to which the
                                          partitions need to be reassigned in
                                          the form "0,1,2". This is required
                                          for automatic topic reassignment.
--execute [execute]                     This option does the actual
                                          reassignment. By default, the tool
                                          does a dry run
--manual-assignment-json-file <manual   The JSON file with the list of manual
  assignment json file path>              reassignments.This option or topics-
                                          to-move-json-file needs to be
                                          specified. The format to use is -
                                        {"partitions":
                                            [{"topic": "foo",
                                              "partition": 1,
                                              "replicas": [1,2,3] }],
                                        "version":1
                                        }
--topics-to-move-json-file <topics to   The JSON file with the list of topics
  reassign json file path>                to reassign.This option or manual-
                                          assignment-json-file needs to be
                                          specified. The format to use is -
                                        {"topics":
                                            [{"topic": "foo"},{"topic": "foo1"}],
                                        "version":1
                                        }
--zookeeper <urls>                      REQUIRED: The connection string for
                                          the zookeeper connection in the form
                                          host:port. Multiple URLS can be
                                          given to allow fail-over.
 

注意此工具預設執行在dry-run(預演)模式下，並不會初始化partiton movement。只有開啟--execute選項，此工具才會開始執行。

叢集擴充套件

在0.8版本中，此工具可以用來擴充套件(expand)一個已存在的叢集。叢集擴充套件涉及到引入了新的broker id。通常當為一個叢集中新增一個broker時，它們不會從現存的topic接收到任何資料，直到使用這個工具把現存的topics/partitions assign到新的broker上。此工具使用2個選項來讓批量轉移topic到新的broker變得更簡單，這2個選項分別為：a)要轉移的topics；b)新增的broker的列表。使用這2個選項，此工具能自動的確定這些topic的partitions的位置。下面的例子把2個topic(foo1，foo2)轉移到新增的broker(id為5，6，7)。

nnarkhed$ ./bin/kafka-reassign-partitions.sh --topics-to-move-json-file topics-to-move.json --broker-list "5,6,7" --execute
 
nnarkhed$ cat topics-to-move.json
{"topics":
     [{"topic": "foo1"},{"topic": "foo2"}],
     "version":1
}

選擇一些partition進行轉移

此工具也可以有選擇的把一些特定的partition的一些replica轉移到一個特定的broker。如果你有一個不負載不均衡的叢集，你可以使用這個工具來有選擇的轉移一些partition。在這種模式下，此工具使用一個檔案，檔案包含了要轉移的partition的列表，還有轉移後replica分佈的broker id列表。

下面的例子把一個partition的replica從broker 1,2,3轉移到了broker 1,2,4上。

nnarkhed$ ./bin/kafka-reassign-partitions.sh --manual-assignment-json-file partitions-to-move.json --execute
 
nnarkhed$ cat partitions-to-move.json
{"partitions":
             [{"topic": "foo",
               "partition": 1,
               "replicas": [1,2,4] }],