private void maybeRevokePartitions() {
        for (String memberId : groupSpec.memberIds()) {
            Map<Uuid, Set<Integer>> oldAssignment = groupSpec.memberAssignment(memberId).partitions();
            Map<Uuid, Set<Integer>> newAssignment = null;

            // The assignor expects to receive the assignment as an immutable map. It leverages
            // this knowledge in order to avoid having to copy all assignments.
            if (!isImmutableMap(oldAssignment)) {
                throw new IllegalStateException("The assignor expect an immutable map.");
            }

            int quota = minimumMemberQuota;
            if (remainingMembersToGetAnExtraPartition > 0) {
                quota++;
                remainingMembersToGetAnExtraPartition--;
            }

            for (Map.Entry<Uuid, Set<Integer>> topicPartitions : oldAssignment.entrySet()) {
                Uuid topicId = topicPartitions.getKey();
                Set<Integer> partitions = topicPartitions.getValue();

                if (subscribedTopicIds.contains(topicId)) {
                    if (partitions.size() <= quota) {
                        quota -= partitions.size();
                    } else {
                        for (Integer partition : partitions) {
                            if (quota > 0) {
                                quota--;
                            } else {
                                if (newAssignment == null) {
                                    // If the new assignment is null, we create a deep copy of the
                                    // original assignment so that we can alter it.
                                    newAssignment = deepCopy(oldAssignment);
                                }
                                // Remove the partition from the new assignment.
                                Set<Integer> parts = newAssignment.get(topicId);
                                parts.remove(partition);
                                if (parts.isEmpty()) {
                                    newAssignment.remove(topicId);
                                }
                                // Add the partition to the unassigned set to be re-assigned later on.
                                unassignedPartitions.add(new TopicIdPartition(topicId, partition));
                            }
                        }
                    }
                } else {
                    if (newAssignment == null) {
                        // If the new assignment is null, we create a deep copy of the
                        // original assignment so that we can alter it.
                        newAssignment = deepCopy(oldAssignment);
                    }
                    // Remove the entire topic.
                    newAssignment.remove(topicId);
                }
            }

            if (quota > 0) {
                unfilledMembers.add(new MemberWithRemainingQuota(memberId, quota));
            }

            if (newAssignment == null) {
                targetAssignment.put(memberId, new MemberAssignmentImpl(oldAssignment));
            } else {
                targetAssignment.put(memberId, new MemberAssignmentImpl(newAssignment));
            }
        }
    }
Revoke the partitions that either are not part of the member's subscriptions or
exceed the maximum quota assigned to each member.

This method ensures that the original assignment is not copied if it is not
altered.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformHomogeneousAssignmentBuilder.java