Xiang Li

I am not sure if Less() could be changed to drop the comparison of leader and compare partition.id only. The sort on leader will have the following benefit, according to...

As a second thought, shall we trigger a re-balance when there is a leader change? Could be a solution for this issue?

If I get it correctly, Kafka Java client does not consider leader and sorts on partition in numeric order. See RangeAssignor in https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/consumer/PartitionAssignor.scala

@gatorsmile @dongjoon-hyun Can you help to review this PR? Thanks!