### Question:
DefaultPartitionGrouper implements PartitionGrouper { public Map<TaskId, Set<TopicPartition>> partitionGroups(Map<Integer, Set<String>> topicGroups, Cluster metadata) { Map<TaskId, Set<TopicPartition>> groups = new HashMap<>(); for (Map.Entry<Integer, Set<String>> entry : topicGroups.entrySet()) { Integer topicGroupId = entry.getKey(); Set<String> topicGroup = entry.getValue(); int maxNumPartitions = maxNumPartitions(metadata, topicGroup); for (int partitionId = 0; partitionId < maxNumPartitions; partitionId++) { Set<TopicPartition> group = new HashSet<>(topicGroup.size()); for (String topic : topicGroup) { List<PartitionInfo> partitions = metadata.partitionsForTopic(topic); if (partitionId < partitions.size()) { group.add(new TopicPartition(topic, partitionId)); } } groups.put(new TaskId(topicGroupId, partitionId), Collections.unmodifiableSet(group)); } } return Collections.unmodifiableMap(groups); }  Map<TaskId, Set<TopicPartition>> partitionGroups(Map<Integer, Set<String>> topicGroups, Cluster metadata);  }### Answer:
@Test public void shouldComputeGroupingForTwoGroups() { final PartitionGrouper grouper = new DefaultPartitionGrouper(); final Map<TaskId, Set<TopicPartition>> expectedPartitionsForTask = new HashMap<>(); final Map<Integer, Set<String>> topicGroups = new HashMap<>(); int topicGroupId = 0; topicGroups.put(topicGroupId, mkSet("topic1")); expectedPartitionsForTask.put(new TaskId(topicGroupId, 0), mkSet(new TopicPartition("topic1", 0))); expectedPartitionsForTask.put(new TaskId(topicGroupId, 1), mkSet(new TopicPartition("topic1", 1))); expectedPartitionsForTask.put(new TaskId(topicGroupId, 2), mkSet(new TopicPartition("topic1", 2))); topicGroups.put(++topicGroupId, mkSet("topic2")); expectedPartitionsForTask.put(new TaskId(topicGroupId, 0), mkSet(new TopicPartition("topic2", 0))); expectedPartitionsForTask.put(new TaskId(topicGroupId, 1), mkSet(new TopicPartition("topic2", 1))); assertEquals(expectedPartitionsForTask, grouper.partitionGroups(topicGroups, metadata)); }
@Test public void shouldComputeGroupingForSingleGroupWithMultipleTopics() { final PartitionGrouper grouper = new DefaultPartitionGrouper(); final Map<TaskId, Set<TopicPartition>> expectedPartitionsForTask = new HashMap<>(); final Map<Integer, Set<String>> topicGroups = new HashMap<>(); final int topicGroupId = 0; topicGroups.put(topicGroupId, mkSet("topic1", "topic2")); expectedPartitionsForTask.put( new TaskId(topicGroupId, 0), mkSet(new TopicPartition("topic1", 0), new TopicPartition("topic2", 0))); expectedPartitionsForTask.put( new TaskId(topicGroupId, 1), mkSet(new TopicPartition("topic1", 1), new TopicPartition("topic2", 1))); expectedPartitionsForTask.put( new TaskId(topicGroupId, 2), mkSet(new TopicPartition("topic1", 2))); assertEquals(expectedPartitionsForTask, grouper.partitionGroups(topicGroups, metadata)); }
@Test public void shouldNotCreateAnyTasksBecauseOneTopicHasUnknownPartitions() { final PartitionGrouper grouper = new DefaultPartitionGrouper(); final Map<TaskId, Set<TopicPartition>> expectedPartitionsForTask = new HashMap<>(); final Map<Integer, Set<String>> topicGroups = new HashMap<>(); final int topicGroupId = 0; topicGroups.put(topicGroupId, mkSet("topic1", "unknownTopic", "topic2")); assertEquals(expectedPartitionsForTask, grouper.partitionGroups(topicGroups, metadata)); }