@Test
public void testTopicGroups() {

    final TopologyBuilder builder = new TopologyBuilder();
    builder.addSource("source-1", "topic-1", "topic-1x");
    builder.addSource("source-2", "topic-2");
    builder.addSource("source-3", "topic-3");
    builder.addSource("source-4", "topic-4");
    builder.addSource("source-5", "topic-5");
    builder.addProcessor("processor-1", new MockProcessorSupplier(), "source-1");
    builder.addProcessor("processor-2", new MockProcessorSupplier(), "source-2", "processor-1");
    builder.copartitionSources(mkList("source-1", "source-2"));
    builder.addProcessor("processor-3", new MockProcessorSupplier(), "source-3", "source-4");
    Map<Integer, TopicsInfo> topicGroups = builder.topicGroups("X");
    Map<Integer, TopicsInfo> expectedTopicGroups = new HashMap<>();
    expectedTopicGroups.put(0, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-1", "topic-1x", "topic-2"), Collections.<String>emptySet(), Collections.<String>emptySet()));
    expectedTopicGroups.put(1, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-3", "topic-4"), Collections.<String>emptySet(), Collections.<String>emptySet()));
    expectedTopicGroups.put(2, new TopicsInfo(Collections.<String>emptySet(), mkSet("topic-5"), Collections.<String>emptySet(), Collections.<String>emptySet()));
    assertEquals(3, topicGroups.size());
    assertEquals(expectedTopicGroups, topicGroups);
    Collection<Set<String>> copartitionGroups = builder.copartitionGroups();
    assertEquals(mkSet(mkSet("topic-1", "topic-1x", "topic-2")), new HashSet<>(copartitionGroups));
}