### Question:
KafkaTopicClientImpl implements KafkaTopicClient { @Override public Map<String, TopicDescription> describeTopics(final Collection<String> topicNames) { try { return executeWithRetries(() -> adminClient.describeTopics(topicNames).all()); } catch (final Exception e) { throw new KafkaResponseGetFailedException("Failed to Describe Kafka Topics " + topicNames, e); } } 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(expected = KafkaResponseGetFailedException.class) public void shouldFailToDescribeTopicsWhenRetriesExpire() { expect(adminClient.listTopics()).andReturn(getEmptyListTopicResult()); expect(adminClient.describeTopics(anyObject())) .andReturn(describeTopicReturningUnknownPartitionException()) .andReturn(describeTopicReturningUnknownPartitionException()) .andReturn(describeTopicReturningUnknownPartitionException()) .andReturn(describeTopicReturningUnknownPartitionException()) .andReturn(describeTopicReturningUnknownPartitionException()); replay(adminClient); KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient); kafkaTopicClient.describeTopics(Collections.singleton(topicName1)); verify(adminClient); }