### Question:
KafkaTopicClientImpl implements KafkaTopicClient { @Override public void deleteTopics(final List<String> topicsToDelete) { if (!isDeleteTopicEnabled) { log.info("Cannot delete topics since 'delete.topic.enable' is false. "); return; } final DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(topicsToDelete); final Map<String, KafkaFuture<Void>> results = deleteTopicsResult.values(); List<String> failList = Lists.newArrayList(); for (final Map.Entry<String, KafkaFuture<Void>> entry : results.entrySet()) { try { entry.getValue().get(30, TimeUnit.SECONDS); } catch (Exception e) { failList.add(entry.getKey()); } } if (!failList.isEmpty()) { throw new KsqlException("Failed to clean up topics: " + failList.stream() .collect(Collectors.joining(","))); } } KafkaTopicClientImpl(final AdminClient adminClient); @Override void createTopic(
      final String topic,
      final int numPartitions,
      final short replicationFactor
  ); @Override void createTopic(
      final String topic,
      final int numPartitions,
      final short replicationFactor,
      final Map<String, ?> configs
  ); @Override boolean isTopicExists(final String topic); @Override Set<String> listTopicNames(); @Override Set<String> listNonInternalTopicNames(); @Override Map<String, TopicDescription> describeTopics(final Collection<String> topicNames); @Override Map<String, String> getTopicConfig(final String topicName); @Override boolean addTopicConfig(final String topicName, final Map<String, ?> overrides); @Override TopicCleanupPolicy getTopicCleanupPolicy(final String topicName); @Override void deleteTopics(final List<String> topicsToDelete); @Override void deleteInternalTopics(final String applicationId); void close();  }### Answer:
@Test public void shouldDeleteTopics() { expect(adminClient.deleteTopics(anyObject())).andReturn(getDeleteTopicsResult()); replay(adminClient); KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient); List<String> topics = Collections.singletonList(topicName2); kafkaTopicClient.deleteTopics(topics); verify(adminClient); }