@Override
  public Set<String> listTopicNames() {
    try {
      return ExecutorUtil.executeWithRetries(
          () -> adminClient.get().listTopics().names().get(),
          ExecutorUtil.RetryBehaviour.ON_RETRYABLE);
    } catch (final Exception e) {
      throw new KafkaResponseGetFailedException("Failed to retrieve Kafka Topic names", e);
    }
  }
@Test
  public void shouldRetryListTopics() {
    // When:
    givenTopicExists("topic1", 1, 1);
    givenTopicExists("topic2", 1, 2);

    when(adminClient.listTopics())
        .thenAnswer(listTopicResult(new NotControllerException("Not Controller")))
        .thenAnswer(listTopicResult());

    // When:
    kafkaTopicClient.listTopicNames();

    // Then:
    verify(adminClient, times(2)).listTopics();
  }