@Override
  public SchemaKStream<?> buildStream(final PlanBuildContext buildContext) {
    final PlanNode source = getSource();
    final SchemaKStream<?> schemaKStream = source.buildStream(buildContext);

    final QueryContext.Stacker contextStacker = buildContext.buildNodeContext(getId().toString());

    return schemaKStream.into(
        ksqlTopic,
        contextStacker,
        getTimestampColumn()
    );
  }
@Test
  public void shouldBuildSourceNode() {
    // When:
    outputNode.buildStream(planBuildContext);

    // Then:
    verify(sourceNode).buildStream(planBuildContext);
  }