@SuppressWarnings("unchecked")
  @Override
  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 CreateSource) {
        final ConfiguredStatement<CreateSource> createStatement =
            (ConfiguredStatement<CreateSource>) statement;
        return (ConfiguredStatement<T>) forCreateStatement(createStatement).orElse(createStatement);
      } else {
        final ConfiguredStatement<CreateAsSelect> createStatement =
            (ConfiguredStatement<CreateAsSelect>) statement;
        return (ConfiguredStatement<T>) forCreateAsStatement(createStatement).orElse(
            createStatement);
      }
    } catch (final KsqlStatementException e) {
      throw e;
    } catch (final KsqlException e) {
      throw new KsqlStatementException(
          ErrorMessageUtil.buildErrorMessage(e),
          statement.getMaskedStatementText(),
          e.getCause());
    }
  }
@Test
  public void shouldInjectForCtStatement() {
    // Given:
    givenKeyAndValueInferenceSupported();

    // When:
    final ConfiguredStatement<CreateTable> result = injector.inject(ctStatement);

    // Then:
    assertThat(result.getStatement().getElements(),
        is(combineElements(INFERRED_KSQL_KEY_SCHEMA_TABLE, INFERRED_KSQL_VALUE_SCHEMA)));
    assertThat(result.getMaskedStatementText(), is(
        "CREATE TABLE `ct` ("
            + "`key` STRING PRIMARY KEY, "
            + "`intField` INTEGER, "
            + "`bigIntField` BIGINT, "
            + "`doubleField` DOUBLE, "
            + "`stringField` STRING, "
            + "`booleanField` BOOLEAN, "
            + "`arrayField` ARRAY<INTEGER>, "
            + "`mapField` MAP<STRING, BIGINT>, "
            + "`structField` STRUCT<`s0` BIGINT>, "
            + "`decimalField` DECIMAL(4, 2)) "
            + "WITH (KAFKA_TOPIC='some-topic', KEY_FORMAT='protobuf', VALUE_FORMAT='avro');"
    ));
  }