public Map<TaskId, Set<TopicPartition>> partitionGroups(final Map<Subtopology, Set<String>> topicGroups, final Cluster metadata) {
        return partitionGroups(topicGroups, new HashMap<>(), new HashMap<>(), metadata);
    }
@Test
    public void shouldComputeGroupingForTwoGroups() {
        final PartitionGrouper grouper = new PartitionGrouper();
        final Map<TaskId, Set<TopicPartition>> expectedPartitionsForTask = new HashMap<>();
        final Map<Subtopology, Set<String>> topicGroups = new HashMap<>();

        topicGroups.put(SUBTOPOLOGY_0, mkSet("topic1"));
        expectedPartitionsForTask.put(new TaskId(SUBTOPOLOGY_0.nodeGroupId, 0, SUBTOPOLOGY_0.namedTopology), mkSet(new TopicPartition("topic1", 0)));
        expectedPartitionsForTask.put(new TaskId(SUBTOPOLOGY_0.nodeGroupId, 1, SUBTOPOLOGY_0.namedTopology), mkSet(new TopicPartition("topic1", 1)));
        expectedPartitionsForTask.put(new TaskId(SUBTOPOLOGY_0.nodeGroupId, 2, SUBTOPOLOGY_0.namedTopology), mkSet(new TopicPartition("topic1", 2)));

        topicGroups.put(SUBTOPOLOGY_1, mkSet("topic2"));
        expectedPartitionsForTask.put(new TaskId(SUBTOPOLOGY_1.nodeGroupId, 0, SUBTOPOLOGY_1.namedTopology), mkSet(new TopicPartition("topic2", 0)));
        expectedPartitionsForTask.put(new TaskId(SUBTOPOLOGY_1.nodeGroupId, 1, SUBTOPOLOGY_1.namedTopology), mkSet(new TopicPartition("topic2", 1)));

        assertEquals(expectedPartitionsForTask, grouper.partitionGroups(topicGroups, metadata));
    }