@Override
  public Map<String, TopicDescription> describeTopics(final Collection<String> topicNames) {
    try {
      return ExecutorUtil.executeWithRetries(
          () -> adminClient.get().describeTopics(
              topicNames,
              new DescribeTopicsOptions().includeAuthorizedOperations(true)
          ).allTopicNames().get(),
          ExecutorUtil.RetryBehaviour.ON_RETRYABLE);
    } catch (final ExecutionException e) {
      throw new KafkaResponseGetFailedException(
          "Failed to Describe Kafka Topic(s): " + topicNames, e.getCause());
    } catch (final TopicAuthorizationException e) {
      throw new KsqlTopicAuthorizationException(
          AclOperation.DESCRIBE, topicNames);
    } catch (final Exception e) {
      throw new KafkaResponseGetFailedException(
          "Failed to Describe Kafka Topic(s): " + topicNames, e);
    }
  }
@Test
  public void shouldRetryDescribeTopicDuringValidateCreateTopicOnRetryableException() {
    // Given:
    givenTopicExists("topicName", 1, 2);
    givenTopicConfigs(
        "topicName",
        overriddenConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "8640000000")
    );

    when(adminClient.describeTopics(anyCollection(), any()))
        .thenAnswer(describeTopicsResult()) // checks that topic exists
        .thenAnswer(describeTopicsResult(new UnknownTopicOrPartitionException("meh"))) // fails during validateProperties
        .thenAnswer(describeTopicsResult()); // succeeds the third time

    // When:
    kafkaTopicClient.validateCreateTopic("topicName", 1, (short) 2, configs);

    // Then:
    verify(adminClient, times(3)).describeTopics(anyCollection(), any());
  }