private boolean allSubscriptionsEqual(Set<String> allTopics,
                                          Map<String, Subscription> subscriptions,
                                          Map<String, List<TopicPartition>> consumerToOwnedPartitions,
                                          Set<TopicPartition> partitionsWithMultiplePreviousOwners) {
        boolean isAllSubscriptionsEqual = true;

        Set<String> subscribedTopics = new HashSet<>();

        // keep track of all previously owned partitions so we can invalidate them if invalid input is
        // detected, eg two consumers somehow claiming the same partition in the same/current generation
        Map<TopicPartition, String> allPreviousPartitionsToOwner = new HashMap<>();

        for (Map.Entry<String, Subscription> subscriptionEntry : subscriptions.entrySet()) {
            final String consumer = subscriptionEntry.getKey();
            final Subscription subscription = subscriptionEntry.getValue();

            // initialize the subscribed topics set if this is the first subscription
            if (subscribedTopics.isEmpty()) {
                subscribedTopics.addAll(subscription.topics());
            } else if (isAllSubscriptionsEqual && !(subscription.topics().size() == subscribedTopics.size()
                && subscribedTopics.containsAll(subscription.topics()))) {
                isAllSubscriptionsEqual = false;
            }

            MemberData memberData = memberData(subscription);
            final int memberGeneration = memberData.generation.orElse(DEFAULT_GENERATION);
            maxGeneration = Math.max(maxGeneration, memberGeneration);

            List<TopicPartition> ownedPartitions = new ArrayList<>();
            consumerToOwnedPartitions.put(consumer, ownedPartitions);

            // the member has a valid generation, so we can consider its owned partitions if it has the highest
            // generation amongst
            for (final TopicPartition tp : memberData.partitions) {
                if (allTopics.contains(tp.topic())) {
                    String otherConsumer = allPreviousPartitionsToOwner.get(tp);
                    if (otherConsumer == null) {
                        // this partition is not owned by other consumer in the same generation
                        ownedPartitions.add(tp);
                        allPreviousPartitionsToOwner.put(tp, consumer);
                    } else {
                        final int otherMemberGeneration = subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION);

                        if (memberGeneration == otherMemberGeneration) {
                            // if two members of the same generation own the same partition, revoke the partition
                            log.error("Found multiple consumers {} and {} claiming the same TopicPartition {} in the "
                                            + "same generation {}, this will be invalidated and removed from their previous assignment.",
                                    consumer, otherConsumer, tp, memberGeneration);
                            partitionsWithMultiplePreviousOwners.add(tp);
                            consumerToOwnedPartitions.get(otherConsumer).remove(tp);
                            allPreviousPartitionsToOwner.put(tp, consumer);
                        } else if (memberGeneration > otherMemberGeneration) {
                            // move partition from the member with an older generation to the member with the newer generation
                            ownedPartitions.add(tp);
                            consumerToOwnedPartitions.get(otherConsumer).remove(tp);
                            allPreviousPartitionsToOwner.put(tp, consumer);
                            log.warn("Consumer {} in generation {} and consumer {} in generation {} claiming the same " +
                                            "TopicPartition {} in different generations. The topic partition will be " +
                                            "assigned to the member with the higher generation {}.",
                                    consumer, memberGeneration,
                                    otherConsumer, otherMemberGeneration,
                                    tp,
                                    memberGeneration);
                        } else {
                            // let the other member continue to own the topic partition
                            log.warn("Consumer {} in generation {} and consumer {} in generation {} claiming the same " +
                                            "TopicPartition {} in different generations. The topic partition will be " +
                                            "assigned to the member with the higher generation {}.",
                                    consumer, memberGeneration,
                                    otherConsumer, otherMemberGeneration,
                                    tp,
                                    otherMemberGeneration);
                        }
                    }
                }
            }
        }
        return isAllSubscriptionsEqual;
    }
Returns true iff all consumers have an identical subscription. Also fills out the passed in
{@code consumerToOwnedPartitions} with each consumer's previously owned and still-subscribed partitions,
and the {@code partitionsWithMultiplePreviousOwners} with any partitions claimed by multiple previous owners
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java