public static  <K> void build(
      final LogicalSchema schema,
      final Formats formats,
      final Optional<TimestampColumn> timestampColumn,
      final String topicName,
      final KStream<K, GenericRow> stream,
      final ExecutionKeyFactory<K> executionKeyFactory,
      final QueryContext queryContext,
      final RuntimeBuildContext buildContext
  ) {
    final PhysicalSchema physicalSchema = PhysicalSchema.from(
        schema,
        formats.getKeyFeatures(),
        formats.getValueFeatures()
    );

    final Serde<K> keySerde = executionKeyFactory.buildKeySerde(
        formats.getKeyFormat(),
        physicalSchema,
        queryContext
    );

    final Serde<GenericRow> valueSerde = buildContext.buildValueSerde(
        formats.getValueFormat(),
        physicalSchema,
        queryContext
    );

    final Optional<TransformTimestamp<K>> tsTransformer = timestampTransformer(
        buildContext,
        queryContext,
        schema,
        timestampColumn
    );

    final KStream<K, GenericRow> transformed = tsTransformer
        .map(t -> stream.transform(t, Named.as(TIMESTAMP_TRANSFORM_NAME
            + StreamsUtil.buildOpName(queryContext))))
        .orElse(stream);

    transformed.to(topicName, Produced.with(keySerde, valueSerde));
  }
@Test
  public void shouldBuildStreamUsingTransformTimestampWhenTimestampIsSpecified() {
    // When
    SinkBuilder.build(
        SCHEMA,
        Formats.of(KEY_FORMAT, VALUE_FORMAT, SerdeFeatures.of(), SerdeFeatures.of()),
        Optional.of(new TimestampColumn(ColumnName.of("BLUE"), Optional.empty())),
        TOPIC,
        kStream,
        executionKeyFactory,
        queryContext,
        buildContext
    );

    // Then
    final InOrder inOrder = Mockito.inOrder(kStream);
    inOrder.verify(kStream).transform(any(), any(Named.class));
    inOrder.verify(kStream).to(eq(TOPIC), any());
    inOrder.verifyNoMoreInteractions();
  }