@Test
public void testBuild() {

    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.addProcessor("processor-3", new MockProcessorSupplier(), "source-3", "source-4");
    ProcessorTopology topology0 = builder.build("X", 0);
    ProcessorTopology topology1 = builder.build("X", 1);
    ProcessorTopology topology2 = builder.build("X", 2);
    assertEquals(mkSet("source-1", "source-2", "processor-1", "processor-2"), nodeNames(topology0.processors()));
    assertEquals(mkSet("source-3", "source-4", "processor-3"), nodeNames(topology1.processors()));
    assertEquals(mkSet("source-5"), nodeNames(topology2.processors()));
}