...

{"version":1,
 "partitions":[{"topic": "foo1",
                "partition": 0,
		        "replicas": [1,2,5]
               },
               {"topic": "foo2",
                "partition": 1,
		        "replicas": [7,9,10]
               }]            
}

Proposed Changes

Reassignment Cancellation

The main idea is support clean, safe cancellation of pending reassignments in /admin/reassign_partitions znode in a timely fashion, and support more reassignments while currently some reassignments are  in-flight. 

...

private def onPartitionReassignment(topicPartition: TopicPartition, reassignedPartitionContext: ReassignedPartitionsContext) {
  if (zkClient.reassignCancelInPlace()) return // if the ReassignCancelZNode exists , skip reassignment

This can serve as a guard against accidentally starting reassignments on some Production clusters which are very performance sensitive due to reassignments.  

New reassignments while existing reassignments in-flight  

In order to support submitting extra reassignments while existing reassignments are still in-flight.  An extra znode /admin/reassign_partitions_extra  which has the same JSON format as /admin/reassign_partitions.   Three extra more options --generate-extra  --verify-extra  --execute-extra  are will be added to  kafka-reassign-partitions.sh.   The controller will be informed of the extra reassignments via a ZooKeeper watch.   It will get all topic/partitions from /admin/reassign_partitions_extra  and  add to /admin/reassign_partitions,  then trigger the reassignments onPartitionReassignment()  of the  topic/partitions.   

...