class CreateSourceFactory {

  public CreateStreamCommand createStreamCommand(
      final CreateStream statement,
      final KsqlConfig ksqlConfig
  ) {
    final SourceName sourceName = statement.getName();
    final CreateSourceProperties props = statement.getProperties();
    final String topicName = ensureTopicExists(props, serviceContext);
    final LogicalSchema schema = buildSchema(statement.getElements());
    final Optional<TimestampColumn> timestampColumn =
        buildTimestampColumn(ksqlConfig, props, schema);

    return new CreateStreamCommand(
        sourceName,
        schema,
        timestampColumn,
        topicName,
        buildFormats(schema, props, ksqlConfig),
        getWindowInfo(props),
        Optional.of(statement.isOrReplace())
    );
  }

  public  CreateSourceFactory(final ServiceContext serviceContext);
  @VisibleForTesting  CreateSourceFactory(
      final ServiceContext serviceContext,
      final SerdeFeaturessSupplier keySerdeFeaturesSupplier,
      final SerdeFeaturessSupplier valueSerdeFeaturesSupplier,
      final KeySerdeFactory keySerdeFactory,
      final ValueSerdeFactory valueSerdeFactory
  );

}