kafka 0.8增加了重新分配partition功能，进行扩容,增加减少replica，移动partition

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

if --topics-to-move-json-file is

used to generate reassignment

configuration

--execute                               Kick off the reassignment as specified

by the --reassignment-json-file

option.

--generate                              Generate a candidate partition

reassignment configuration. Note

that this only generates a candidate

assignment, it does not execute it.

--reassignment-json-file <manual        The JSON file with the partition

assignment json file path>              reassignment configurationThe format

to use is -

{"partitions":

[{"topic": "foo",

"partition": 1,

"replicas": [1,2,3] }],

"version":1

}

--topics-to-move-json-file <topics to   Generate a reassignment configuration

reassign json file path>                to move the partitions of the

specified topics to the list of

brokers specified by the --broker-

list option. The format to use is -

{"topics":

[{"topic": "foo"},{"topic": "foo1"}],

"version":1

}

--verify                                Verify if the reassignment completed

as specified by the --reassignment-

json-file option.

--zookeeper <urls>                      REQUIRED: The connection string for

the zookeeper connection in the form

host:port. Multiple URLS can be

given to allow fail-over.

narkhed$ cat partitions-to-move.json

{"partitions":

[{"topic": "foo",

"partition": 1,

"replicas": [1,2,3] }],

"version":1

}

bin/kafka-reassign-partitions.sh --topics-to-move-json-file dirk_topic_move.json  --zookeeper host34/kafka-dirktest --broker-list "0341,0361" --generate

生成将topic dirkzhang放到341和361上的json，用json执行execute就可以了

Current partition replica assignment

{"version":1,"partitions":[{"topic":"dirkzhang","partition":6,"replicas":[361,371]},{"topic":"dirkzhang","partition":0,"replicas":[361,371]},{"topic":"dirkzhang","partition":5,"replicas":[341,371]},{"topic":"dirkzhang","partition":4,"replicas":[371,361]},{"topic":"dirkzhang","partition":3,"replicas":[361,341]},{"topic":"dirkzhang","partition":7,"replicas":[371,341]},{"topic":"dirkzhang","partition":1,"replicas":[371,341]},{"topic":"dirkzhang","partition":2,"replicas":[341,361]}]}

Proposed partition reassignment configuration

{"version":1,"partitions":[{"topic":"dirkzhang","partition":6,"replicas":[361,341]},{"topic":"dirkzhang","partition":5,"replicas":[341,361]},{"topic":"dirkzhang","partition":0,"replicas":[361,341]},{"topic":"dirkzhang","partition":4,"replicas":[361,341]},{"topic":"dirkzhang","partition":3,"replicas":[341,361]},{"topic":"dirkzhang","partition":1,"replicas":[341,361]},{"topic":"dirkzhang","partition":7,"replicas":[341,361]},{"topic":"dirkzhang","partition":2,"replicas":[361,341]}]}

-----------------------------------------------------------------

bin/kafka-reassign-partitions.sh --zookeeper host34/kafka-dirktest --reassignment-json-file dirk_move_partition.json --execute

将partition0,1的replica放到341,371上（brokerid前边不能有零，是int）

dirk_move_partition.json:

{"version":1,"partitions":

[{"topic":"dirkzhang",

"partition":0,

"replicas":[341,371]

},

{"topic":"dirkzhang",

"partition":1,

"replicas":[341,371]

}]

}

------------------------------------------------

用verify检查是否迁移成功

bin/kafka-reassign-partitions.sh --reassignment-json-file dirk_move_partition.json  --zookeeper host34/kafka-dirktest --verify

Status of partition reassignment:

Reassignment of partition [dirkzhang,1] completed successfully

----------------------------------------------------

收集replica迁移，选leader的日志

bin/kafka-run-class.sh kafka.tools.StateChangeLogMerger

--end-time <end timestamp in the        The latest timestamp of state change

format java.text.                       log entries to be merged (default:

SimpleDateFormat@f17a63e7>              9999-12-31 23:59:59,999)

--logs <file1,file2,...>                Comma separated list of state change

logs or a regex for the log file

names

--logs-regex <for example: /tmp/state-  Regex to match the state change log

change.log*>                            files to be merged

--partitions <0,1,2,...>                Comma separated list of partition ids

whose state change logs should be

merged

--start-time <start timestamp in the    The earliest timestamp of state change

format java.text.                       log entries to be merged (default:

SimpleDateFormat@f17a63e7>              0000-00-00 00:00:00,000)

--topic <topic>                         The topic whose state change logs

should be merged

bin/kafka-run-class.sh kafka.tools.StateChangeLogMerger --topic dirkzhang --partition 0,1,2 --start-time "2015-07-02 00:00:00,000" --end-time "2015-07-03 00:00:00,000" --logs /data/kafka_2.10-0.8.2.1/logs/state-change.log