@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 shouldBuildOutputNodeForInsertIntoAvroFromNonAvro() {
    // Given:
    givenInsertIntoNode();

    KeyFormat.nonWindowed(
        FormatInfo.of(
            FormatFactory.AVRO.name(),
            ImmutableMap.of(ConnectProperties.FULL_SCHEMA_NAME, "key-name")
        ),
        SerdeFeatures.of()
    );

    ValueFormat.of(
        FormatInfo.of(
            FormatFactory.AVRO.name(),
            ImmutableMap.of(ConnectProperties.FULL_SCHEMA_NAME, "name")
        ),
        SerdeFeatures.of()
    );

    // When:
    outputNode.buildStream(planBuildContext);

    // Then:
    verify(sourceStream).into(eq(ksqlTopic), any(), any());
  }