@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 shouldThrowIfCtasKeyTableElementsNotCompatibleWrongKeyType() {
    // Given:
    givenFormatsAndProps("protobuf", null,
        ImmutableMap.of("KEY_SCHEMA_ID", new IntegerLiteral(42)));
    givenDDLSchemaAndFormats(LOGICAL_SCHEMA_INT_KEY, "protobuf", "avro",
        SerdeFeature.WRAP_SINGLES, SerdeFeature.UNWRAP_SINGLES);

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> injector.inject(ctasStatement)
    );

    // Then:
    assertThat(e.getMessage(),
        containsString("The following key columns are changed, missing or reordered: "
            + "[`key` INTEGER KEY]. Schema from schema registry is [`key` STRING KEY]"));
  }