@Override
    public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, Subscription> subscriptions) {
        Map<String, List<TopicPartition>> assignment = new HashMap();
        Iterator subIter = subscriptions.keySet().iterator();

        //initialize subscription mappings for assignment
        while (subIter.hasNext()) {
            String memberId = (String) subIter.next();
            assignment.put(memberId, new ArrayList());
        }

        ArrayList<String> consumerList = new ArrayList(Utils.sorted(subscriptions.keySet()));
        Iterator partIter = this.allPartitionsSorted(partitionsPerTopic, subscriptions).iterator();

        //assign partitions at random
        while (partIter.hasNext()) {
            TopicPartition partition = (TopicPartition) partIter.next();
            String topic = partition.topic();

            int rand = ThreadLocalRandom.current().nextInt(0, consumerList.size());
            while (!((Subscription) subscriptions.get(consumerList.get(rand))).topics().contains(topic)) {
                rand = ThreadLocalRandom.current().nextInt(0, consumerList.size());
            }

            (assignment.get(consumerList.get(rand))).add(partition);
        }

        return assignment;
    }
@Test
    public void assignmentWorksWithMultipleTopics() {
        String topic1 = "testTopic1";
        String topic2 = "testTopic2";
        String topic3 = "testTopic3";
        List<String> topicList = new ArrayList<String>();
        topicList.add(topic1);
        topicList.add(topic2);
        topicList.add(topic3);
        String consumerId = "testConsumer";
        Map<String, Integer> partitionsPerTopic = new HashMap<>();
        partitionsPerTopic.put(topic1, 1);
        partitionsPerTopic.put(topic2, 1);
        partitionsPerTopic.put(topic3, 1);

        Map<String, List<TopicPartition>> assignment = testAssignor.assign(partitionsPerTopic,
            Collections.singletonMap(consumerId, new Subscription(topicList)));

        List<TopicPartition> testAssignment = new ArrayList<>();
        testAssignment.add(new TopicPartition(topic1, 0));
        testAssignment.add(new TopicPartition(topic2, 0));
        testAssignment.add(new TopicPartition(topic3, 0));

        assertEquals(testAssignment, assignment.get(consumerId));
    }