private void assignOwnedPartitions() {

            for (Map.Entry<String, List<TopicPartition>> consumerEntry : currentAssignment.entrySet()) {
                String consumer = consumerEntry.getKey();
                List<TopicPartition> ownedPartitions = consumerEntry.getValue().stream()
                        .filter(tp -> {
                            boolean mismatch = rackInfo.racksMismatch(consumer, tp);
                            if (mismatch) {
                                maybeRevokedPartitions.put(tp, consumer);
                            }
                            return !mismatch;
                        })
                        .sorted(Comparator.comparing(TopicPartition::partition).thenComparing(TopicPartition::topic))
                        .collect(Collectors.toList());

                List<TopicPartition> consumerAssignment = assignment.get(consumer);

                for (TopicPartition doublyClaimedPartition : partitionsWithMultiplePreviousOwners) {
                    if (ownedPartitions.contains(doublyClaimedPartition)) {
                        log.error("Found partition {} still claimed as owned by consumer {}, despite being claimed by multiple "
                                        + "consumers already in the same generation. Removing it from the ownedPartitions",
                                doublyClaimedPartition, consumer);
                        ownedPartitions.remove(doublyClaimedPartition);
                    }
                }

                if (ownedPartitions.size() < minQuota) {
                    // the expected assignment size is more than this consumer has now, so keep all the owned partitions
                    // and put this member into the unfilled member list
                    if (ownedPartitions.size() > 0) {
                        consumerAssignment.addAll(ownedPartitions);
                        assignedPartitions.addAll(ownedPartitions);
                    }
                    unfilledMembersWithUnderMinQuotaPartitions.add(consumer);
                } else if (ownedPartitions.size() >= maxQuota && currentNumMembersWithOverMinQuotaPartitions < expectedNumMembersWithOverMinQuotaPartitions) {
                    // consumer owned the "maxQuota" of partitions or more, and we're still under the number of expected members
                    // with more than the minQuota partitions, so keep "maxQuota" of the owned partitions, and revoke the rest of the partitions
                    currentNumMembersWithOverMinQuotaPartitions++;
                    if (currentNumMembersWithOverMinQuotaPartitions == expectedNumMembersWithOverMinQuotaPartitions) {
                        unfilledMembersWithExactlyMinQuotaPartitions.clear();
                    }
                    List<TopicPartition> maxQuotaPartitions = ownedPartitions.subList(0, maxQuota);
                    consumerAssignment.addAll(maxQuotaPartitions);
                    assignedPartitions.addAll(maxQuotaPartitions);
                    for (TopicPartition topicPartition : ownedPartitions.subList(maxQuota, ownedPartitions.size())) {
                        maybeRevokedPartitions.put(topicPartition, consumer);
                    }
                } else {
                    // consumer owned at least "minQuota" of partitions
                    // so keep "minQuota" of the owned partitions, and revoke the rest of the partitions
                    List<TopicPartition> minQuotaPartitions = ownedPartitions.subList(0, minQuota);
                    consumerAssignment.addAll(minQuotaPartitions);
                    assignedPartitions.addAll(minQuotaPartitions);
                    for (TopicPartition topicPartition : ownedPartitions.subList(minQuota, ownedPartitions.size())) {
                        maybeRevokedPartitions.put(topicPartition, consumer);
                    }
                    // this consumer is potential maxQuota candidate since we're still under the number of expected members
                    // with more than the minQuota partitions. Note, if the number of expected members with more than
                    // the minQuota partitions is 0, it means minQuota == maxQuota, and there are no potentially unfilled
                    if (currentNumMembersWithOverMinQuotaPartitions < expectedNumMembersWithOverMinQuotaPartitions) {
                        unfilledMembersWithExactlyMinQuotaPartitions.add(consumer);
                    }
                }
            }
        }
Constructs a constrained assignment builder.

@param partitionsPerTopic                   The partitions for each subscribed topic
@param rackInfo                             Rack information for consumers and racks
@param consumerToOwnedPartitions            Each consumer's previously owned and still-subscribed partitions
@param partitionsWithMultiplePreviousOwners The partitions being claimed in the previous assignment of multiple consumers
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java