private void checkValidityAndBalance(
        Map<String, MemberSubscriptionAndAssignmentImpl> memberSubscriptionSpec,
        GroupAssignment computedGroupAssignment
    ) {
        List<String> membersList = new ArrayList<>(computedGroupAssignment.members().keySet());
        int numMembers = membersList.size();
        List<Integer> totalAssignmentSizesOfAllMembers = new ArrayList<>(membersList.size());
        membersList.forEach(member -> {
            Map<Uuid, Set<Integer>> computedAssignmentForMember = computedGroupAssignment
                .members().get(member).partitions();
            int sum = computedAssignmentForMember.values().stream().mapToInt(Set::size).sum();
            totalAssignmentSizesOfAllMembers.add(sum);
        });

        for (int i = 0; i < numMembers; i++) {
            String memberId = membersList.get(i);
            Map<Uuid, Set<Integer>> computedAssignmentForMember =
                computedGroupAssignment.members().get(memberId).partitions();
            // Each member is subscribed to topics of all the partitions assigned to it.
            computedAssignmentForMember.keySet().forEach(topicId -> {
                // Check if the topic exists in the subscription.
                assertTrue(memberSubscriptionSpec.get(memberId).subscribedTopicIds().contains(topicId),
                    "Error: Partitions for topic " + topicId + " are assigned to member " + memberId +
                        " but it is not part of the members subscription ");
            });

            for (int j = i + 1; j < numMembers; j++) {
                String otherMemberId = membersList.get(j);
                Map<Uuid, Set<Integer>> computedAssignmentForOtherMember = computedGroupAssignment
                    .members().get(otherMemberId).partitions();
                // Each partition should be assigned to at most one member
                computedAssignmentForMember.keySet().forEach(topicId -> {
                    Set<Integer> intersection = new HashSet<>();
                    if (computedAssignmentForOtherMember.containsKey(topicId)) {
                        intersection = new HashSet<>(computedAssignmentForMember.get(topicId));
                        intersection.retainAll(computedAssignmentForOtherMember.get(topicId));
                    }
                    assertTrue(
                        intersection.isEmpty(),
                        "Error : Member 1 " + memberId + " and Member 2 " + otherMemberId +
                            "have common partitions assigned to them " + computedAssignmentForOtherMember.get(topicId)
                    );
                });

                // Difference in the sizes of any two partitions should be 1 at max
                int size1 = totalAssignmentSizesOfAllMembers.get(i);
                int size2 = totalAssignmentSizesOfAllMembers.get(j);
                assertTrue(
                    Math.abs(size1 - size2) <= 1,
                    "Size of one assignment is greater than the other assignment by more than one partition "
                        + size1 + " " + size2 + "abs = " + Math.abs(size1 - size2)
                );
            }
        }
    }
Verifies that the given assignment is valid with respect to the given subscriptions.
Validity requirements:
     - each member is subscribed to topics of all partitions assigned to it, and
     - each partition is assigned to no more than one member.
Balance requirements:
     - the assignment is fully balanced (the numbers of topic partitions assigned to memberSubscriptionSpec differ by at most one), or
     - there is no topic partition that can be moved from one member to another with 2+ fewer topic partitions.

@param memberSubscriptionSpec        Members subscription metadata structure from the group Spec.
@param computedGroupAssignment       Assignment computed by the uniform assignor.
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java