public CreateTableCommand createTableCommand(
      final KsqlStructuredDataOutputNode outputNode,
      final Optional<RefinementInfo> emitStrategy
  ) {
    Optional<WindowInfo> windowInfo =
        outputNode.getKsqlTopic().getKeyFormat().getWindowInfo();

    if (windowInfo.isPresent() && emitStrategy.isPresent()) {
      final WindowInfo info = windowInfo.get();
      windowInfo = Optional.of(WindowInfo.of(
          info.getType(),
          info.getSize(),
          Optional.of(emitStrategy.get().getOutputRefinement())
      ));
    }

    return new CreateTableCommand(
        outputNode.getSinkName().get(),
        outputNode.getSchema(),
        outputNode.getTimestampColumn(),
        outputNode.getKsqlTopic().getKafkaTopicName(),
        Formats.from(outputNode.getKsqlTopic()),
        windowInfo,
        Optional.of(outputNode.getOrReplace()),
        Optional.of(false)
    );
  }
@Test
  public void shouldCreateValueSerdeToValidateValueFormatCanHandleValueSchema() {
    // Given:
    givenCommandFactoriesWithMocks();
    final CreateTable statement = new CreateTable(SOME_NAME, TABLE_ELEMENTS, false, true,
        withProperties, false);

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

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

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