@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 shouldCallInto() {
    // When:
    final SchemaKStream<?> result = outputNode.buildStream(planBuildContext);

    // Then:
    verify(sourceStream).into(
        eq(ksqlTopic),
        stackerCaptor.capture(),
        eq(outputNode.getTimestampColumn())
    );
    assertThat(
        stackerCaptor.getValue().getQueryContext().getContext(),
        equalTo(ImmutableList.of("0"))
    );
    assertThat(result, sameInstance(sinkStream));
  }