RangeAssignor extends AbstractPartitionAssignor { @Override public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, List<String>> subscriptions) { Map<String, List<String>> consumersPerTopic = consumersPerTopic(subscriptions); Map<String, List<TopicPartition>> assignment = new HashMap<>(); for (String memberId : subscriptions.keySet()) assignment.put(memberId, new ArrayList<TopicPartition>()); for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) { String topic = topicEntry.getKey(); List<String> consumersForTopic = topicEntry.getValue(); Integer numPartitionsForTopic = partitionsPerTopic.get(topic); if (numPartitionsForTopic == null) { continue; } Collections.sort(consumersForTopic); int numConsumersForConsumer = consumersForTopic.size(); int N_numPartitionsPerConsumer = numPartitionsForTopic / numConsumersForConsumer; int M_consumersWithExtraPartition = numPartitionsForTopic % numConsumersForConsumer; List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic); for (int i = 0, n = consumersForTopic.size(); i < n; i++) { int start = N_numPartitionsPerConsumer * i + Math.min(i, M_consumersWithExtraPartition); int length = N_numPartitionsPerConsumer + (i + 1 > M_consumersWithExtraPartition ? 0 : 1); assignment.get(consumersForTopic.get(i)) .addAll(partitions.subList(start, start + length)); } } return assignment; }  @Override String name(); @Override Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer> partitionsPerTopic,
        Map<String/* memberId */, List<String/* topic */>> subscriptions);  }
@Test public void testOneConsumerNoTopic() { String consumerId = "consumer"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, Collections.singletonMap(consumerId, Collections.<String>emptyList())); assertEquals(Collections.singleton(consumerId), assignment.keySet()); assertTrue(assignment.get(consumerId).isEmpty()); }
@Test public void testOneConsumerNonexistentTopic() { String topic = "topic"; String consumerId = "consumer"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, Collections.singletonMap(consumerId, Arrays.asList(topic))); assertEquals(Collections.singleton(consumerId), assignment.keySet()); assertTrue(assignment.get(consumerId).isEmpty()); }
@Test public void testOneConsumerOneTopic() { String topic = "topic"; String consumerId = "consumer"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); partitionsPerTopic.put(topic, 3); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, Collections.singletonMap(consumerId, Arrays.asList(topic))); assertEquals(Collections.singleton(consumerId), assignment.keySet()); assertAssignment(Arrays.asList( new TopicPartition(topic, 0), new TopicPartition(topic, 1), new TopicPartition(topic, 2)), assignment.get(consumerId)); }
@Test public void testOnlyAssignsPartitionsFromSubscribedTopics() { String topic = "topic"; String otherTopic = "other"; String consumerId = "consumer"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); partitionsPerTopic.put(topic, 3); partitionsPerTopic.put(otherTopic, 3); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, Collections.singletonMap(consumerId, Arrays.asList(topic))); assertEquals(Collections.singleton(consumerId), assignment.keySet()); assertAssignment(Arrays.asList( new TopicPartition(topic, 0), new TopicPartition(topic, 1), new TopicPartition(topic, 2)), assignment.get(consumerId)); }
@Test public void testOneConsumerMultipleTopics() { String topic1 = "topic1"; String topic2 = "topic2"; String consumerId = "consumer"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); partitionsPerTopic.put(topic1, 1); partitionsPerTopic.put(topic2, 2); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, Collections.singletonMap(consumerId, Arrays.asList(topic1, topic2))); assertEquals(Collections.singleton(consumerId), assignment.keySet()); assertAssignment(Arrays.asList( new TopicPartition(topic1, 0), new TopicPartition(topic2, 0), new TopicPartition(topic2, 1)), assignment.get(consumerId)); }
@Test public void testTwoConsumersOneTopicOnePartition() { String topic = "topic"; String consumer1 = "consumer1"; String consumer2 = "consumer2"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); partitionsPerTopic.put(topic, 1); Map<String, List<String>> consumers = new HashMap<>(); consumers.put(consumer1, Arrays.asList(topic)); consumers.put(consumer2, Arrays.asList(topic)); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, consumers); assertAssignment(Arrays.asList(new TopicPartition(topic, 0)), assignment.get(consumer1)); assertAssignment(Collections.<TopicPartition>emptyList(), assignment.get(consumer2)); }
@Test public void testTwoConsumersOneTopicTwoPartitions() { String topic = "topic"; String consumer1 = "consumer1"; String consumer2 = "consumer2"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); partitionsPerTopic.put(topic, 2); Map<String, List<String>> consumers = new HashMap<>(); consumers.put(consumer1, Arrays.asList(topic)); consumers.put(consumer2, Arrays.asList(topic)); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, consumers); assertAssignment(Arrays.asList(new TopicPartition(topic, 0)), assignment.get(consumer1)); assertAssignment(Arrays.asList(new TopicPartition(topic, 1)), assignment.get(consumer2)); }
@Test public void testMultipleConsumersMixedTopics() { String topic1 = "topic1"; String topic2 = "topic2"; String consumer1 = "consumer1"; String consumer2 = "consumer2"; String consumer3 = "consumer3"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); partitionsPerTopic.put(topic1, 3); partitionsPerTopic.put(topic2, 2); Map<String, List<String>> consumers = new HashMap<>(); consumers.put(consumer1, Arrays.asList(topic1)); consumers.put(consumer2, Arrays.asList(topic1, topic2)); consumers.put(consumer3, Arrays.asList(topic1)); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, consumers); assertAssignment(Arrays.asList( new TopicPartition(topic1, 0)), assignment.get(consumer1)); assertAssignment(Arrays.asList( new TopicPartition(topic1, 1), new TopicPartition(topic2, 0), new TopicPartition(topic2, 1)), assignment.get(consumer2)); assertAssignment(Arrays.asList( new TopicPartition(topic1, 2)), assignment.get(consumer3)); }
@Test public void testTwoConsumersTwoTopicsSixPartitions() { String topic1 = "topic1"; String topic2 = "topic2"; String consumer1 = "consumer1"; String consumer2 = "consumer2"; Map<String, Integer> partitionsPerTopic = new HashMap<>(); partitionsPerTopic.put(topic1, 3); partitionsPerTopic.put(topic2, 3); Map<String, List<String>> consumers = new HashMap<>(); consumers.put(consumer1, Arrays.asList(topic1, topic2)); consumers.put(consumer2, Arrays.asList(topic1, topic2)); Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, consumers); assertAssignment(Arrays.asList( new TopicPartition(topic1, 0), new TopicPartition(topic1, 1), new TopicPartition(topic2, 0), new TopicPartition(topic2, 1)), assignment.get(consumer1)); assertAssignment(Arrays.asList( new TopicPartition(topic1, 2), new TopicPartition(topic2, 2)), assignment.get(consumer2)); }