@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 assignmentWorksWithMultipleConsumers() {
        String topic = "testTopic";
        List<String> topicList = new ArrayList<String>();
        topicList.add(topic);
        String consumerId1 = "testConsumer1";
        String consumerId2 = "testConsumer2";
        Map<String, Integer> partitionsPerTopic = new HashMap<>();
        partitionsPerTopic.put(topic, 1);
        Map<String, Subscription> consumers = new HashMap<>();
        consumers.put(consumerId1, new Subscription(topicList));
        consumers.put(consumerId2, new Subscription(topicList));

        Map<String, List<TopicPartition>> assignment = testAssignor.assign(partitionsPerTopic, consumers);
        List<TopicPartition> testAssignment = new ArrayList<>();
        testAssignment.add(new TopicPartition(topic, 0));

        if (assignment.get(consumerId1).equals(Collections.<String>emptyList()) && assignment.get(consumerId2).equals(testAssignment)) {
            return;
        } else if (assignment.get(consumerId1).equals(testAssignment) && assignment.get(consumerId2).equals(Collections.<String>emptyList())) {
            return;
        } else {
            fail("Partition should be assigned to a single consumer");
        }
    }