def verify_validity_and_balance(subscriptions, assignment):
    """
    Verifies that the given assignment is valid with respect to the given subscriptions
    Validity requirements:
    - each consumer is subscribed to topics of all partitions assigned to it, and
    - each partition is assigned to no more than one consumer
    Balance requirements:
    - the assignment is fully balanced (the numbers of topic partitions assigned to consumers differ by at most one), or
    - there is no topic partition that can be moved from one consumer to another with 2+ fewer topic partitions

    :param subscriptions  topic subscriptions of each consumer
    :param assignment: given assignment for balance check
    """
    assert six.viewkeys(subscriptions) == six.viewkeys(assignment)

    consumers = sorted(six.viewkeys(assignment))
    for i in range(len(consumers)):
        consumer = consumers[i]
        partitions = assignment[consumer].partitions()
        for partition in partitions:
            assert partition.topic in subscriptions[consumer], (
                'Error: Partition {} is assigned to consumer {}, '
                'but it is not subscribed to topic {}\n'
                'Subscriptions: {}\n'
                'Assignments: {}'.format(partition, consumers[i], partition.topic, subscriptions, assignment)
            )
        if i == len(consumers) - 1:
            continue

        for j in range(i + 1, len(consumers)):
            other_consumer = consumers[j]
            other_partitions = assignment[other_consumer].partitions()
            partitions_intersection = set(partitions).intersection(set(other_partitions))
            assert partitions_intersection == set(), (
                'Error: Consumers {} and {} have common partitions '
                'assigned to them: {}\n'
                'Subscriptions: {}\n'
                'Assignments: {}'.format(consumer, other_consumer, partitions_intersection, subscriptions, assignment)
            )

            if abs(len(partitions) - len(other_partitions)) <= 1:
                continue

            assignments_by_topic = group_partitions_by_topic(partitions)
            other_assignments_by_topic = group_partitions_by_topic(other_partitions)
            if len(partitions) > len(other_partitions):
                for topic in six.iterkeys(assignments_by_topic):
                    assert topic not in other_assignments_by_topic, (
                        'Error: Some partitions can be moved from {} ({} partitions) '
                        'to {} ({} partitions) '
                        'to achieve a better balance\n'
                        'Subscriptions: {}\n'
                        'Assignments: {}'.format(consumer, len(partitions), other_consumer, len(other_partitions), subscriptions, assignment)
                    )
            if len(other_partitions) > len(partitions):
                for topic in six.iterkeys(other_assignments_by_topic):
                    assert topic not in assignments_by_topic, (
                        'Error: Some partitions can be moved from {} ({} partitions) '
                        'to {} ({} partitions) '
                        'to achieve a better balance\n'
                        'Subscriptions: {}\n'
                        'Assignments: {}'.format(other_consumer, len(other_partitions), consumer, len(partitions), subscriptions, assignment)
                    )

    Verifies that the given assignment is valid with respect to the given subscriptions
    Validity requirements:
    - each consumer is subscribed to topics of all partitions assigned to it, and
    - each partition is assigned to no more than one consumer
    Balance requirements:
    - the assignment is fully balanced (the numbers of topic partitions assigned to consumers differ by at most one), or
    - there is no topic partition that can be moved from one consumer to another with 2+ fewer topic partitions

    :param subscriptions  topic subscriptions of each consumer
    :param assignment: given assignment for balance check
    
test/test_assignors.py