public CreateStreamCommand createStreamCommand(final KsqlStructuredDataOutputNode outputNode) {
    return new CreateStreamCommand(
        outputNode.getSinkName().get(),
        outputNode.getSchema(),
        outputNode.getTimestampColumn(),
        outputNode.getKsqlTopic().getKafkaTopicName(),
        Formats.from(outputNode.getKsqlTopic()),
        outputNode.getKsqlTopic().getKeyFormat().getWindowInfo(),
        Optional.of(outputNode.getOrReplace()),
        Optional.of(false)
    );
  }
@Test
  public void shouldBuildSerdeFeaturesForStream() {
    // Given:
    givenCommandFactoriesWithMocks();
    final CreateStream statement = new CreateStream(SOME_NAME, ONE_KEY_ONE_VALUE, false, true,
        withProperties, false);
    final LogicalSchema schema = LogicalSchema.builder()
        .keyColumn(ColumnName.of("k"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("bob"), SqlTypes.STRING)
        .build();
    when(keyOptionsSupplier.build(any(), any(), any(), any()))
        .thenReturn(SerdeFeatures.of(SerdeFeature.WRAP_SINGLES));
    when(valOptionsSupplier.build(any(), any(), any(), any()))
        .thenReturn(SerdeFeatures.of(SerdeFeature.UNWRAP_SINGLES));

    // When:
    final CreateStreamCommand cmd = createSourceFactory.createStreamCommand(
        statement,
        ksqlConfig
    );

    // Then:
    verify(keyOptionsSupplier).build(
        schema,
        FormatFactory.of(SourcePropertiesUtil.getKeyFormat(statement.getProperties(), SOME_NAME)),
        SerdeFeatures.of(),
        ksqlConfig
    );
    verify(valOptionsSupplier).build(
        schema,
        FormatFactory.of(SourcePropertiesUtil.getValueFormat(statement.getProperties())),
        statement.getProperties().getValueSerdeFeatures(),
        ksqlConfig
    );
    assertThat(cmd.getFormats().getKeyFeatures(), is(SerdeFeatures.of(SerdeFeature.WRAP_SINGLES)));
    assertThat(cmd.getFormats().getValueFeatures(),
        is(SerdeFeatures.of(SerdeFeature.UNWRAP_SINGLES)));
  }