@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 shouldRegisterKeyOverrideSchemaAvro()
      throws IOException, RestClientException {
    // Given:
    when(schemaRegistryClient.register(anyString(), any(ParsedSchema.class))).thenReturn(1);
    final SchemaAndId schemaAndId = SchemaAndId.schemaAndId(SCHEMA.value(), AVRO_SCHEMA, 1);
    givenStatement("CREATE STREAM source (id int key, f1 varchar) "
        + "WITH ("
        + "kafka_topic='expectedName', "
        + "key_format='AVRO', "
        + "value_format='JSON', "
        + "key_schema_id=1, "
        + "partitions=1"
        + ");", Pair.of(schemaAndId, null));

    // When:
    injector.inject(statement);

    // Then:
    verify(schemaRegistryClient).register("expectedName-key", AVRO_SCHEMA);
  }