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 shouldValidateKeyFormatCanHandleKeySchema() {
    // Given:
    givenCommandFactoriesWithMocks();
    final CreateStream statement = new CreateStream(SOME_NAME, STREAM_ELEMENTS, false, true,
        withProperties, false);

    when(keySerdeFactory.create(
        FormatInfo.of(KAFKA.name()),
        PersistenceSchema.from(EXPECTED_SCHEMA.key(), SerdeFeatures.of()),
        ksqlConfig,
        serviceContext.getSchemaRegistryClientFactory(),
        "",
        NoopProcessingLogContext.INSTANCE,
        Optional.empty()
    )).thenThrow(new RuntimeException("Boom!"));

    // When:
    final Exception e = assertThrows(
        Exception.class,
        () -> createSourceFactory.createStreamCommand(statement, ksqlConfig)
    );

    // Then:
    assertThat(e.getMessage(), containsString("Boom!"));
  }