@ParameterizedTest(name = TEST_NAME_WITH_RACK_CONFIG)
    @EnumSource(RackConfig.class)
    public void testTopicBalanceAfterReassignment(RackConfig rackConfig) {
        initializeRacks(rackConfig);
        List<String> allTopics = topics(topic1, topic2);
        Map<String, List<PartitionInfo>> partitionsPerTopic = new HashMap<>();
        partitionsPerTopic.put(topic1, partitionInfos(topic1, 12));
        partitionsPerTopic.put(topic2, partitionInfos(topic2, 12));
        subscriptions.put(consumer1, subscription(allTopics, 0));
        Map<String, List<TopicPartition>> assignment = assignor.assignPartitions(partitionsPerTopic, subscriptions);

        verifyValidityAndBalance(subscriptions, assignment, partitionsPerTopic);
        assignment.forEach((consumer, tps) -> assertEquals(12, tps.stream().filter(tp -> tp.topic().equals(topic1)).count()));
        assignment.forEach((consumer, tps) -> assertEquals(12, tps.stream().filter(tp -> tp.topic().equals(topic2)).count()));
        assertTrue(assignor.partitionsTransferringOwnership.isEmpty());
        assertTrue(isFullyBalanced(assignment));

        // Add another consumer
        subscriptions.put(consumer1, buildSubscriptionV2Above(allTopics, assignment.get(consumer1), generationId, 0));
        subscriptions.put(consumer2, buildSubscriptionV2Above(allTopics, Collections.emptyList(), generationId, 1));
        assignment = assignor.assignPartitions(partitionsPerTopic, subscriptions);

        verifyValidityAndBalance(subscriptions, assignment, partitionsPerTopic);
        assignment.forEach((consumer, tps) -> assertEquals(6, tps.stream().filter(tp -> tp.topic().equals(topic1)).count()));
        assignment.forEach((consumer, tps) -> assertEquals(6, tps.stream().filter(tp -> tp.topic().equals(topic2)).count()));
        assertTrue(isFullyBalanced(assignment));

        // Add two more consumers
        subscriptions.put(consumer1, buildSubscriptionV2Above(allTopics, assignment.get(consumer1), generationId, 0));
        subscriptions.put(consumer2, buildSubscriptionV2Above(allTopics, assignment.get(consumer2), generationId, 1));
        subscriptions.put(consumer3, buildSubscriptionV2Above(allTopics, Collections.emptyList(), generationId, 2));
        subscriptions.put(consumer4, buildSubscriptionV2Above(allTopics, Collections.emptyList(), generationId, 3));
        assignment = assignor.assignPartitions(partitionsPerTopic, subscriptions);

        verifyValidityAndBalance(subscriptions, assignment, partitionsPerTopic);
        assignment.forEach((consumer, tps) -> assertEquals(3, tps.stream().filter(tp -> tp.topic().equals(topic1)).count()));
        assignment.forEach((consumer, tps) -> assertEquals(3, tps.stream().filter(tp -> tp.topic().equals(topic2)).count()));
        assertTrue(isFullyBalanced(assignment));

        // remove 2 consumers
        subscriptions.remove(consumer1);
        subscriptions.remove(consumer2);
        subscriptions.put(consumer3, buildSubscriptionV2Above(allTopics, assignment.get(consumer3), generationId, 2));
        subscriptions.put(consumer4, buildSubscriptionV2Above(allTopics, assignment.get(consumer4), generationId, 3));
        assignment = assignor.assignPartitions(partitionsPerTopic, subscriptions);

        verifyValidityAndBalance(subscriptions, assignment, partitionsPerTopic);
        assignment.forEach((consumer, tps) -> assertEquals(6, tps.stream().filter(tp -> tp.topic().equals(topic1)).count()));
        assignment.forEach((consumer, tps) -> assertEquals(6, tps.stream().filter(tp -> tp.topic().equals(topic2)).count()));
        assertTrue(assignor.partitionsTransferringOwnership.isEmpty());
        assertTrue(isFullyBalanced(assignment));
    }
This unit test is testing all consumers owned less than minQuota partitions situation
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java