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

    try {
      if (statement.getStatement() instanceof CreateAsSelect) {
        return (ConfiguredStatement<T>) injectForCreateAsSelect(
            (ConfiguredStatement<? extends CreateAsSelect>) statement);
      } else {
        return (ConfiguredStatement<T>) injectForCreateSource(
            (ConfiguredStatement<? extends CreateSource>) statement);
      }
    } catch (final KsqlStatementException e) {
      throw e;
    } catch (final KsqlException e) {
      throw new KsqlStatementException(
          ErrorMessageUtil.buildErrorMessage(e),
          statement.getMaskedStatementText(),
          e.getCause());
    }
  }
@Test
  public void shouldInjectForCreateAsSelect() {
    // When:
    final ConfiguredStatement<CreateAsSelect> configured = injector.inject(csasStatement);

    // Then:
    assertThat(configured.getStatement(), is(csasWithUnwrapping));
  }