@Override
  public boolean isTopicExists(final String topic) {
    LOG.trace("Checking for existence of topic '{}'", topic);
    try {
      ExecutorUtil.executeWithRetries(
          () -> adminClient.get().describeTopics(
              ImmutableList.of(topic),
              new DescribeTopicsOptions().includeAuthorizedOperations(true)
          ).topicNameValues().get(topic).get(),
          RetryBehaviour.ON_RETRYABLE.and(e -> !(e instanceof UnknownTopicOrPartitionException))
      );
      return true;
    } catch (final TopicAuthorizationException e) {
      throw new KsqlTopicAuthorizationException(
          AclOperation.DESCRIBE, Collections.singleton(topic));
    } catch (final Exception e) {
      if (Throwables.getRootCause(e) instanceof UnknownTopicOrPartitionException) {
        return false;
      }

      throw new KafkaResponseGetFailedException("Failed to check if exists for topic: " + topic, e);
    }
  }
@Test
  public void shouldNotListAllTopicsWhenCallingIsTopicExists() {
    // Given
    givenTopicExists("foobar", 1, 1);

    // When
    kafkaTopicClient.isTopicExists("foobar");

    // Then
    verify(adminClient, never()).listTopics();
  }