@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 shouldThrowAuthorizationException() throws Exception {
    // Given:
    givenStatement("CREATE STREAM sink WITH(value_format='AVRO') AS SELECT * FROM SOURCE;");
    when(schemaRegistryClient.register(anyString(), any(ParsedSchema.class)))
        .thenThrow(new RestClientException(
            "User is denied operation Write on Subject", 403, 40301));

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

    // Then:
    assertThat(e.getMessage(), equalTo(
        "Authorization denied to Write on Schema Registry subject: [SINK-key]"));
  }