@Override
  public DdlCommand create(
      final String sqlExpression,
      final DdlStatement ddlStatement,
      final SessionConfig config
  ) {
    return FACTORIES
        .getOrDefault(ddlStatement.getClass(), (statement, cf, ci) -> {
          throw new KsqlException(
              "Unable to find ddl command factory for statement:"
                  + statement.getClass()
                  + " valid statements:"
                  + FACTORIES.keySet()
          );
        })
        .handle(
            this,
            new CallInfo(sqlExpression, config),
            ddlStatement);
  }
@Test
  public void shouldCreateCommandForStreamWithOverriddenProperties() {
    // Given:
    final CreateStream statement = new CreateStream(SOME_NAME, SOME_ELEMENTS, false, true, withProperties, false);

    // When:
    commandFactories.create(sqlExpression, statement, SessionConfig.of(ksqlConfig, OVERRIDES));

    verify(createSourceFactory).createStreamCommand(
        statement,
        ksqlConfig.cloneWithPropertyOverwrite(OVERRIDES));
  }