private boolean performReassignments(List<TopicPartition> reassignablePartitions,
                                             Map<TopicPartition, ConsumerGenerationPair> prevAssignment) {
            boolean reassignmentPerformed = false;
            boolean modified;

            // repeat reassignment until no partition can be moved to improve the balance
            do {
                modified = false;
                // reassign all reassignable partitions (starting from the partition with least potential consumers and if needed)
                // until the full list is processed or a balance is achieved
                Iterator<TopicPartition> partitionIterator = reassignablePartitions.iterator();
                while (partitionIterator.hasNext() && !isBalanced()) {
                    TopicPartition partition = partitionIterator.next();

                    // the partition must have at least two consumers
                    if (topic2AllPotentialConsumers.get(partition.topic()).size() <= 1)
                        log.error("Expected more than one potential consumer for partition '{}'", partition);

                    // the partition must have a current consumer
                    String consumer = currentPartitionConsumer.get(partition);
                    if (consumer == null)
                        log.error("Expected partition '{}' to be assigned to a consumer", partition);

                    if (prevAssignment.containsKey(partition) &&
                            currentAssignment.get(consumer).size() > currentAssignment.get(prevAssignment.get(partition).consumer).size() + 1) {
                        reassignPartition(partition, prevAssignment.get(partition).consumer);
                        reassignmentPerformed = true;
                        modified = true;
                        continue;
                    }

                    // check if a better-suited consumer exist for the partition; if so, reassign it
                    // Use consumer within rack if possible
                    String consumerRack = rackInfo.consumerRacks.get(consumer);
                    Set<String> partitionRacks = rackInfo.partitionRacks.get(partition);
                    boolean foundRackConsumer = false;
                    if (consumerRack != null && !partitionRacks.isEmpty() && partitionRacks.contains(consumerRack)) {
                        for (String otherConsumer : topic2AllPotentialConsumers.get(partition.topic())) {
                            String otherConsumerRack = rackInfo.consumerRacks.get(otherConsumer);
                            if (otherConsumerRack == null || !partitionRacks.contains(otherConsumerRack))
                                continue;
                            if (currentAssignment.get(consumer).size() > currentAssignment.get(otherConsumer).size() + 1) {
                                reassignPartition(partition);
                                reassignmentPerformed = true;
                                modified = true;
                                foundRackConsumer = true;
                                break;
                            }
                        }
                    }
                    if (!foundRackConsumer) {
                        for (String otherConsumer : topic2AllPotentialConsumers.get(partition.topic())) {
                            if (currentAssignment.get(consumer).size() > currentAssignment.get(otherConsumer).size() + 1) {
                                reassignPartition(partition);
                                reassignmentPerformed = true;
                                modified = true;
                                break;
                            }
                        }
                    }
                }
            } while (modified);

            return reassignmentPerformed;
        }
Balance the current assignment using the data structures created in the assignPartitions(...) method above.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java