@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 shouldThrowOnDescribeOnTopicAuthorizationException() {
    // Given:
    when(adminClient.describeTopics(anyCollection(), any()))
        .thenAnswer(describeTopicsResult(new TopicAuthorizationException("meh")));

    // When:
    final Exception e = assertThrows(
        KsqlTopicAuthorizationException.class,
        () -> kafkaTopicClient.describeTopics(ImmutableList.of("topic1"))
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "Authorization denied to Describe on topic(s): [topic1]"));
  }