@Override
  public void deleteTopics(final Collection<String> topicsToDelete) {
    if (topicsToDelete.isEmpty()) {
      return;
    }

    final DeleteTopicsResult deleteTopicsResult = adminClient.get().deleteTopics(topicsToDelete);
    final Map<String, KafkaFuture<Void>> results = deleteTopicsResult.topicNameValues();
    final List<String> failList = Lists.newArrayList();
    final List<Pair<String, Throwable>> exceptionList = Lists.newArrayList();
    for (final Map.Entry<String, KafkaFuture<Void>> entry : results.entrySet()) {
      try {
        entry.getValue().get(30, TimeUnit.SECONDS);
      } catch (final Exception e) {
        final Throwable rootCause = ExceptionUtils.getRootCause(e);

        if (rootCause instanceof TopicDeletionDisabledException) {
          throw new TopicDeletionDisabledException("Topic deletion is disabled. "
              + "To delete the topic, you must set '" + DELETE_TOPIC_ENABLE + "' to true in "
              + "the Kafka broker configuration.");
        } else if (rootCause instanceof TopicAuthorizationException) {
          throw new KsqlTopicAuthorizationException(
              AclOperation.DELETE, Collections.singleton(entry.getKey()));
        } else if (!(rootCause instanceof UnknownTopicOrPartitionException)) {
          LOG.error(String.format("Could not delete topic '%s'", entry.getKey()), e);
          failList.add(entry.getKey());
          exceptionList.add(new Pair<>(entry.getKey(), rootCause));
        }
      }
    }

    if (!failList.isEmpty()) {
      throw new KafkaDeleteTopicsException("Failed to clean up topics: "
          + String.join(",", failList), exceptionList);
    }
  }
@Test
  @SuppressWarnings("unchecked")
  public void shouldNotThrowKafkaDeleteTopicsExceptionWhenMissingTopic() {
    // Given:
    when(adminClient.deleteTopics(any(Collection.class)))
        .thenAnswer(deleteTopicsResult(new UnknownTopicOrPartitionException("error")));

    // When:
    kafkaTopicClient.deleteTopics(ImmutableList.of("aTopic"));

    // Then: did NOT throw.
  }