public static StatementExecutorResponse execute(
      final ConfiguredStatement<ListTopics> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    final KafkaTopicClient client = serviceContext.getTopicClient();
    final Map<String, TopicDescription> topicDescriptions = listTopics(client, statement);

    if (statement.getStatement().getShowExtended()) {
      final KafkaConsumerGroupClient consumerGroupClient
          = new KafkaConsumerGroupClientImpl(serviceContext::getAdminClient);
      final Map<String, List<Integer>> topicConsumersAndGroupCount
          = getTopicConsumerAndGroupCounts(consumerGroupClient);

      final List<KafkaTopicInfoExtended> topicInfoExtendedList = topicDescriptions.values()
          .stream().map(desc ->
              topicDescriptionToTopicInfoExtended(desc, topicConsumersAndGroupCount))
          .collect(Collectors.toList());

      return StatementExecutorResponse.handled(Optional.of(
          new KafkaTopicsListExtended(statement.getMaskedStatementText(), topicInfoExtendedList)));
    } else {
      final List<KafkaTopicInfo> topicInfoList = topicDescriptions.values()
          .stream().map(ListTopicsExecutor::topicDescriptionToTopicInfo)
          .collect(Collectors.toList());

      return StatementExecutorResponse.handled(Optional.of(
          new KafkaTopicsList(statement.getMaskedStatementText(), topicInfoList)));
    }
  }
@Test
  public void shouldListKafkaTopicsWithoutInternalTopics() {
    // Given:
    engine.givenKafkaTopic("topic1");
    engine.givenKafkaTopic("topic2");
    engine.givenKafkaTopic("_confluent_any_topic");

    // When:
    final KafkaTopicsList topicsList =
        (KafkaTopicsList) CustomExecutors.LIST_TOPICS.execute(
            engine.configure("LIST TOPICS;"),
            mock(SessionProperties.class),
            engine.getEngine(),
            serviceContext
        ).getEntity().orElseThrow(IllegalStateException::new);

    // Then:
    assertThat(topicsList.getTopics(), containsInAnyOrder(
        new KafkaTopicInfo("topic1", ImmutableList.of(1)),
        new KafkaTopicInfo("topic2", ImmutableList.of(1))
    ));
  }