The partition reassignment tool can be used to move some topics off of the current set of brokers to the newly added brokers. This is typically useful while expanding an existing cluster since it is easier to move entire topics to the new set of brokers, than moving one partition at a time. When used to do this, the user should provide a list of topics that should be moved to the new set of brokers and a target list of new brokers. The tool then evenly distributes all partitions for the given list of topics across the new set of brokers. During this move, the replication factor of the topic is kept constant. Effectively the replicas for all partitions for the input list of topics are moved from the old set of brokers to the newly added brokers.
For instance, the following example will move all partitions for topics foo1,foo2 to the new set of brokers 5,6. At the end of this move, all partitions for topics foo1 and foo2 will only exist on brokers 5,6.
Since the tool accepts the input list of topics as a json file, you first need to identify the topics you want to move and create the json file as follows:
|
1
2
3
4
5
|
>
cat
topics-to-move.json
{
"topics"
: [{
"topic"
:
"foo1"
},
{
"topic"
:
"foo2"
}],
"version"
:1
}
|
Once the json file is ready, use the partition reassignment tool to generate a candidate assignment:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
> bin
/kafka-reassign-partitions
.sh --zookeeper localhost:2181 --topics-to-move-json-
file
topics-to-move.json --broker-list
"5,6"
--generate
Current partition replica assignment
{
"version"
:1,
"partitions"
:[{
"topic"
:
"foo1"
,
"partition"
:2,
"replicas"
:[1,2]},
{
"topic"
:
"foo1"
,
"partition"
:0,
"replicas"
:[3,4]},
{
"topic"
:
"foo2"
,
"partition"
:2,
"replicas"
:[1,2]},
{
"topic"
:
"foo2"
,
"partition"
:0,
"replicas"
:[3,4]},
{
"topic"
:
"foo1"
,
"partition"
:1,
"replicas"
:[2,3]},
{
"topic"
:
"foo2"
,
"partition"
:1,
"replicas"
:[2,3]}]
}
Proposed partition reassignment configuration
{
"version"
:1,
"partitions"
:[{
"topic"
:
"foo1"
,
"partition"
:2,
"replicas"
:[5,6]},
{
"topic"
:
"foo1"
,
"partition"
:0,
"replicas"
:[5,6]},
{
"topic"
:
"foo2"
,
"partition"
:2,
"replicas"
:[5,6]},
{
"topic"
:
"foo2"
,
"partition"
:0,
"replicas"
:[5,6]},
{
"topic"
:
"foo1"
,
"partition"
:1,
"replicas"
:[5,6]},
{
"topic"
:
"foo2"
,
"partition"
:1,
"replicas"
:[5,6]}]
}
|
The tool generates a candidate assignment that will move all partitions from topics foo1,foo2 to brokers 5,6. Note, however, that at this point, the partition movement has not started, it merely tells you the current assignment and the proposed new assignment. The current assignment should be saved in case you want to rollback to it. The new assignment should be saved in a json file (e.g. expand-cluster-reassignment.json) to be input to the tool with the --execute option as follows:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
> bin
/kafka-reassign-partitions
.sh --zookeeper localhost:2181 --reassignment-json-
file
expand
-cluster-reassignment.json --execute
Current partition replica assignment
{
"version"
:1,
"partitions"
:[{
"topic"
:
"foo1"
,
"partition"
:2,
"replicas"
:[1,2]},
{
"topic"
:
"foo1"
,
"partition"
:0,
"replicas"
:[3,4]},
{
"topic"
:
"foo2"
,
"partition"
:2,
"replicas"
:[1,2]},
{
"topic"
:
"foo2"
,
"partition"
:0,
"replicas"
:[3,4]},
{
"topic"
:
"foo1"
,
"partition"
:1,
"replicas"
:[2,3]},
{
"topic"
:
"foo2"
,
"partition"
:1,
"replicas"
:[2,3]}]
}
Save this to use as the --reassignment-json-
file
option during rollback
Successfully started reassignment of partitions
{
"version"
:1,
"partitions"
:[{
"topic"
:
"foo1"
,
"partition"
:2,
"replicas"
:[5,6]},
{
"topic"
:
"foo1"
,
"partition"
:0,
"replicas"
:[5,6]},
{
"topic"
:
"foo2"
,
"partition"
:2,
"replicas"
:[5,6]},
{
"topic"
:
"foo2"
,
"partition"
:0,
"replicas"
:[5,6]},
{
"topic"
:
"foo1"
,
"partition"
:1,
"replicas"
:[5,6]},
{
"topic"
:
"foo2"
,
"partition"
:1,
"replicas"
:[5,6]}]
}
|
Finally, the --verify option can be used with the tool to check the status of the partition reassignment. Note that the same expand-cluster-reassignment.json (used with the --execute option) should be used with the --verify option:
|
1
2
3
4
5
6
7
8
|
> bin
/kafka-reassign-partitions
.sh --zookeeper localhost:2181 --reassignment-json-
file
expand
-cluster-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [foo1,0] completed successfully
Reassignment of partition [foo1,1] is
in
progress
Reassignment of partition [foo1,2] is
in
progress
Reassignment of partition [foo2,0] completed successfully
Reassignment of partition [foo2,1] completed successfully
Reassignment of partition [foo2,2] completed successfully
|
注意：
The partition reassignment tool does not have the ability to automatically generate a reassignment plan for decommissioning brokers yet. As such, the admin has to come up with a reassignment plan to move the replica for all partitions hosted on the broker to be decommissioned, to the rest of the brokers. This can be relatively tedious as the reassignment needs to ensure that all the replicas are not moved from the decommissioned broker to only one other broker. To make this process effortless, we plan to add tooling support for decommissioning brokers in the future.