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 shouldFailToAssertWrongConfigs() {
    // Given:
    final Map<String, Literal> configs = ImmutableMap.of(
        "partitions", new IntegerLiteral(10), "replicas", new IntegerLiteral(10), "abc", new IntegerLiteral(23));
    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 KsqlRestException e = assertThrows(KsqlRestException.class,
        () -> AssertTopicExecutor.execute(statement, mock(SessionProperties.class), engine, serviceContext));

    // Then:
    assertThat(e.getResponse().getStatus(), is(417));
    assertThat(((KsqlErrorMessage) e.getResponse().getEntity()).getMessage(), is("Mismatched configuration for topic topicName: For config partitions, expected 10 got 1\n"
        + "Mismatched configuration for topic topicName: For config replicas, expected 10 got 0\n"
        + "Cannot assert unknown topic property: abc"));
  }