public static StatementExecutorResponse execute(
      final ConfiguredStatement<AssertTopic> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    return AssertExecutor.execute(statement.getMaskedStatementText(),
        statement.getStatement(),
        executionContext.getKsqlConfig().getInt(KSQL_ASSERT_TOPIC_DEFAULT_TIMEOUT_MS),
        serviceContext,
        (stmt, sc) -> assertTopic(
            sc.getTopicClient(),
            ((AssertTopic) stmt).getTopic(),
            ((AssertTopic) stmt).getConfig(),
            stmt.checkExists()),
        (str, stmt) -> new AssertTopicEntity(
            str,
            ((AssertTopic) stmt).getTopic(),
            stmt.checkExists())
    );
  }
@Test
  public void shouldAssertTopic() {
    // Given:
    final Map<String, Literal> configs = ImmutableMap.of(
        "partitions", new IntegerLiteral(1), "replicas", new IntegerLiteral(0));
    final AssertTopic assertTopic = new AssertTopic(Optional.empty(), "topicName", configs, Optional.empty(), true);
    final ConfiguredStatement<AssertTopic> statement = ConfiguredStatement
        .of(KsqlParser.PreparedStatement.of("", assertTopic),
            SessionConfig.of(ksqlConfig, ImmutableMap.of()));

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

    // Then:
    assertThat("expected response!", entity.isPresent());
    assertThat(((AssertTopicEntity) entity.get()).getTopicName(), is("topicName"));
    assertThat(((AssertTopicEntity) entity.get()).getExists(), is(true));
  }