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 shouldAssertNotExistSchemaById() {
    // Given
    final AssertSchema assertSchema = new AssertSchema(Optional.empty(), Optional.empty(), Optional.of(100), Optional.empty(), false);
    final ConfiguredStatement<AssertSchema> statement = ConfiguredStatement
        .of(KsqlParser.PreparedStatement.of("", assertSchema),
            SessionConfig.of(ksqlConfig, ImmutableMap.of()));

    // When:
    final Optional<KsqlEntity> entity = AssertSchemaExecutor
        .execute(statement, mock(SessionProperties.class), engine, serviceContext).getEntity();

    // Then:
    assertThat("expected response!", entity.isPresent());
    assertThat(((AssertSchemaEntity) entity.get()).getSubject(), is(Optional.empty()));
    assertThat(((AssertSchemaEntity) entity.get()).getId(), is(Optional.of(100)));
    assertThat(((AssertSchemaEntity) entity.get()).getExists(), is(false));
  }