public static StatementExecutorResponse execute(
      final ConfiguredStatement<AssertSchema> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    return AssertExecutor.execute(
        statement.getMaskedStatementText(),
        statement.getStatement(),
        executionContext.getKsqlConfig().getInt(KSQL_ASSERT_SCHEMA_DEFAULT_TIMEOUT_MS),
        serviceContext,
        (stmt, sc) -> assertSchema(
            sc.getSchemaRegistryClient(),
            ((AssertSchema) stmt).getSubject(),
            ((AssertSchema) stmt).getId(),
            stmt.checkExists()),
        (str, stmt) -> new AssertSchemaEntity(
            str,
            ((AssertSchema) stmt).getSubject(),
            ((AssertSchema) stmt).getId(),
            stmt.checkExists())
    );
  }
@Test
  public void shouldFailToAssertNotExistSchemaBySubjectAndId() {
    // Given
    final AssertSchema assertSchema = new AssertSchema(Optional.empty(), Optional.of("abc"), Optional.of(500), Optional.empty(), false);
    final ConfiguredStatement<AssertSchema> statement = ConfiguredStatement
        .of(KsqlParser.PreparedStatement.of("", assertSchema),
            SessionConfig.of(ksqlConfig, ImmutableMap.of()));

    // When:
    final KsqlRestException e = assertThrows(KsqlRestException.class, () ->
        AssertSchemaExecutor.execute(statement, mock(SessionProperties.class), engine, serviceContext));

    // Then:
    assertThat(e.getResponse().getStatus(), is(417));
    assertThat(((KsqlErrorMessage) e.getResponse().getEntity()).getMessage(), is("Schema with subject name abc id 500 exists"));
  }