@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); assertEquals(Arrays.asList( new TopicPartition(topic1, 0), new TopicPartition(topic1, 2), new TopicPartition(topic2, 1)), assignment.get(consumer1)); assertEquals(Arrays.asList( new TopicPartition(topic1, 1), new TopicPartition(topic2, 0), new TopicPartition(topic2, 2)), assignment.get(consumer2)); }
@Override public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, List<String>> subscriptions) { Map<String, List<TopicPartition>> assignment = new HashMap<>(); for (String memberId : subscriptions.keySet()) assignment.put(memberId, new ArrayList<TopicPartition>()); CircularIterator<String> assigner = new CircularIterator<>(Utils.sorted(subscriptions.keySet())); for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) { final String topic = partition.topic(); while (!subscriptions.get(assigner.peek()) .contains(topic)) { assigner.next(); } assignment.get(assigner.next()) .add(partition); } return assignment; }
RoundRobinAssignor extends AbstractPartitionAssignor { @Override public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, List<String>> subscriptions) { Map<String, List<TopicPartition>> assignment = new HashMap<>(); for (String memberId : subscriptions.keySet()) assignment.put(memberId, new ArrayList<TopicPartition>()); CircularIterator<String> assigner = new CircularIterator<>(Utils.sorted(subscriptions.keySet())); for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) { final String topic = partition.topic(); while (!subscriptions.get(assigner.peek()) .contains(topic)) { assigner.next(); } assignment.get(assigner.next()) .add(partition); } return assignment; } }
RoundRobinAssignor extends AbstractPartitionAssignor { @Override public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, List<String>> subscriptions) { Map<String, List<TopicPartition>> assignment = new HashMap<>(); for (String memberId : subscriptions.keySet()) assignment.put(memberId, new ArrayList<TopicPartition>()); CircularIterator<String> assigner = new CircularIterator<>(Utils.sorted(subscriptions.keySet())); for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) { final String topic = partition.topic(); while (!subscriptions.get(assigner.peek()) .contains(topic)) { assigner.next(); } assignment.get(assigner.next()) .add(partition); } return assignment; }  }
RoundRobinAssignor extends AbstractPartitionAssignor { @Override public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, List<String>> subscriptions) { Map<String, List<TopicPartition>> assignment = new HashMap<>(); for (String memberId : subscriptions.keySet()) assignment.put(memberId, new ArrayList<TopicPartition>()); CircularIterator<String> assigner = new CircularIterator<>(Utils.sorted(subscriptions.keySet())); for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) { final String topic = partition.topic(); while (!subscriptions.get(assigner.peek()) .contains(topic)) { assigner.next(); } assignment.get(assigner.next()) .add(partition); } return assignment; }  @Override Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer/*  */> partitionsPerTopic,
        Map<String/* memberId */, List<String/* topic */>> subscriptions); List<TopicPartition> allPartitionsSorted(Map<String, Integer> partitionsPerTopic,
        Map<String, List<String>> subscriptions); @Override String name(); }
RoundRobinAssignor extends AbstractPartitionAssignor { @Override public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, List<String>> subscriptions) { Map<String, List<TopicPartition>> assignment = new HashMap<>(); for (String memberId : subscriptions.keySet()) assignment.put(memberId, new ArrayList<TopicPartition>()); CircularIterator<String> assigner = new CircularIterator<>(Utils.sorted(subscriptions.keySet())); for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) { final String topic = partition.topic(); while (!subscriptions.get(assigner.peek()) .contains(topic)) { assigner.next(); } assignment.get(assigner.next()) .add(partition); } return assignment; }  @Override Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer/*  */> partitionsPerTopic,
        Map<String/* memberId */, List<String/* topic */>> subscriptions); List<TopicPartition> allPartitionsSorted(Map<String, Integer> partitionsPerTopic,
        Map<String, List<String>> subscriptions); @Override String name();  }