@SuppressWarnings("unchecked")
  @Override
  public <T extends Statement> ConfiguredStatement<T> inject(
      final ConfiguredStatement<T> statement
  ) {
    try {
      if (statement.getStatement() instanceof CreateAsSelect) {
        registerForCreateAs((ConfiguredStatement<? extends CreateAsSelect>) statement);
      } else if (statement.getStatement() instanceof CreateSource) {
        registerForCreateSource((ConfiguredStatement<? extends CreateSource>) statement);
      }
    } catch (final KsqlStatementException e) {
      throw e;
    } catch (final KsqlException e) {
      throw new KsqlStatementException(
          ErrorMessageUtil.buildErrorMessage(e),
          statement.getMaskedStatementText(),
          e.getCause());
    }
    // Remove schema id from SessionConfig
    return stripSchemaIdConfig(statement);
  }
@Test
  public void shouldSupportPrimitiveValueSchemasInCreateAsStmts() throws Exception {
    // Given:
    givenStatement("CREATE STREAM sink "
        + "WITH(value_format='AVRO', wrap_single_value='false') AS "
        + "SELECT * FROM SOURCE;");

    when(valFeatures.enabled(SerdeFeature.UNWRAP_SINGLES)).thenReturn(true);

    // When:
    injector.inject(statement);

    // Then:
    verify(schemaRegistryClient).register("SINK-value", AVRO_UNWRAPPED_VALUE_SCHEMA);
  }