class DefaultSchemaInjector implements Injector {

  @SuppressWarnings("unchecked")
  @Override
  public <T extends Statement> ConfiguredStatement<T> inject(
      final ConfiguredStatement<T> statement
  ) {
    if (!(statement.getStatement() instanceof CreateSource)) {
      return statement;
    }

    final ConfiguredStatement<CreateSource> createStatement =
        (ConfiguredStatement<CreateSource>) statement;

    try {
      return (ConfiguredStatement<T>) forCreateStatement(createStatement).orElse(createStatement);
    } catch (final KsqlStatementException e) {
      throw e;
    } catch (final KsqlException e) {
      throw new KsqlStatementException(
          ErrorMessageUtil.buildErrorMessage(e), 
          statement.getStatementText(), 
          e.getCause());
    }
  }

  public  DefaultSchemaInjector(final TopicSchemaSupplier schemaSupplier);

}