@Override
  public SchemaKStream<?> buildStream(final PlanBuildContext buildContext) {
    final Stacker contextStacker = buildContext.buildNodeContext(getId().toString());
    return schemaKStreamFactory.create(
        buildContext,
        dataSource,
        contextStacker.push(SOURCE_OP_NAME)
    );
  }
@Test
  public void shouldBuildSourceNode() {
    // When:
    realStream = buildStream(node);

    // Then:
    final TopologyDescription.Source node = (TopologyDescription.Source) getNodeByName(realBuilder.build(), PlanTestUtil.SOURCE_NODE);
    final List<String> successors = node.successors().stream().map(TopologyDescription.Node::name).collect(Collectors.toList());
    assertThat(node.predecessors(), equalTo(Collections.emptySet()));
    assertThat(successors, equalTo(Collections.singletonList(PlanTestUtil.TRANSFORM_NODE)));
    assertThat(node.topicSet(), equalTo(ImmutableSet.of("topic")));
  }