@Override
    public Map<String, List<TopicPartition>> assignPartitions(Map<String, List<PartitionInfo>> partitionsPerTopic,
                                                              Map<String, Subscription> subscriptions) {
        Map<String, List<TopicPartition>> consumerToOwnedPartitions = new HashMap<>();
        Set<TopicPartition> partitionsWithMultiplePreviousOwners = new HashSet<>();

        List<PartitionInfo> allPartitions = new ArrayList<>();
        partitionsPerTopic.values().forEach(allPartitions::addAll);
        RackInfo rackInfo = new RackInfo(allPartitions, subscriptions);

        AbstractAssignmentBuilder assignmentBuilder;
        if (allSubscriptionsEqual(partitionsPerTopic.keySet(), subscriptions, consumerToOwnedPartitions, partitionsWithMultiplePreviousOwners)) {
            log.debug("Detected that all consumers were subscribed to same set of topics, invoking the "
                          + "optimized assignment algorithm");
            partitionsTransferringOwnership = new HashMap<>();
            assignmentBuilder = new ConstrainedAssignmentBuilder(partitionsPerTopic, rackInfo, consumerToOwnedPartitions, partitionsWithMultiplePreviousOwners);
        } else {
            log.debug("Detected that not all consumers were subscribed to same set of topics, falling back to the "
                          + "general case assignment algorithm");
            // we must set this to null for the general case so the cooperative assignor knows to compute it from scratch
            partitionsTransferringOwnership = null;
            assignmentBuilder = new GeneralAssignmentBuilder(partitionsPerTopic, rackInfo, consumerToOwnedPartitions, subscriptions);
        }
        return assignmentBuilder.build();
    }
@Test
    public void testSubscriptionNotEqualAndAssignSamePartitionWith3Generation() {
        Map<String, List<PartitionInfo>> partitionsPerTopic = new HashMap<>();
        partitionsPerTopic.put(topic, partitionInfos(topic, 6));
        partitionsPerTopic.put(topic1, partitionInfos(topic1, 1));
        int[][] sequence = new int[][]{{1, 2, 3}, {1, 3, 2}, {2, 1, 3}, {2, 3, 1}, {3, 1, 2}, {3, 2, 1}};
        for (int[] ints : sequence) {
            subscriptions.put(
                    consumer1,
                    buildSubscriptionV2Above(topics(topic),
                    partitions(tp(topic, 0), tp(topic, 2)), ints[0], 0)
            );
            subscriptions.put(
                    consumer2,
                    buildSubscriptionV2Above(topics(topic),
                    partitions(tp(topic, 1), tp(topic, 2), tp(topic, 3)), ints[1], 1)
            );
            subscriptions.put(
                    consumer3,
                    buildSubscriptionV2Above(topics(topic),
                    partitions(tp(topic, 2), tp(topic, 4), tp(topic, 5)), ints[2], 2)
            );
            subscriptions.put(
                    consumer4,
                    buildSubscriptionV2Above(topics(topic1),
                    partitions(tp(topic1, 0)), 2, 3)
            );

            Map<String, List<TopicPartition>> assign = assignor.assignPartitions(partitionsPerTopic, subscriptions);
            assertEquals(assign.values().stream().mapToInt(List::size).sum(),
                    assign.values().stream().flatMap(List::stream).collect(Collectors.toSet()).size());
            for (List<TopicPartition> list: assign.values()) {
                assertTrue(list.size() >= 1 && list.size() <= 2);
            }
        }
    }