@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 ConsumerEmptyWithoutTopic() {
        String consumerId = "testConsumer";
        Map<String, Integer> partitionsPerTopic = new HashMap<>();
        Map<String, List<TopicPartition>> assignment = testAssignor.assign(
            partitionsPerTopic,
            Collections.singletonMap(consumerId, new Subscription(Collections.<String>emptyList()))
        );

        assertEquals(Collections.singleton(consumerId), assignment.keySet());
        assertTrue(assignment.get(consumerId).isEmpty());
    }