kafka Reassign Partitions Tool

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