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 shouldAllowNonStringKeyColumn() {
    // Given:
    final CreateStream statement = new CreateStream(
        SOME_NAME,
        TableElements.of(tableElement("k", new Type(SqlTypes.INTEGER), KEY_CONSTRAINT)),
        false,
        true,
        withProperties,
        false
    );

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

    // Then:
    assertThat(cmd.getSchema().key(), contains(
        keyColumn(ColumnName.of("k"), SqlTypes.INTEGER)
    ));
  }