kafka Reassign Partitions Tool

swsrl

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