### Question:
KafkaTopicClientImpl implements KafkaTopicClient { @Override public Map<String, String> getTopicConfig(final String topicName) { return topicConfig(topicName, true); } 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 shouldGetTopicConfig() { expect(adminClient.describeConfigs(topicConfigsRequest("fred"))) .andReturn(topicConfigResponse( "fred", overriddenConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "12345"), defaultConfigEntry(TopicConfig.COMPRESSION_TYPE_CONFIG, "snappy") )); replay(adminClient); KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient); final Map<String, String> config = kafkaTopicClient.getTopicConfig("fred"); assertThat(config.get(TopicConfig.RETENTION_MS_CONFIG), is("12345")); assertThat(config.get(TopicConfig.COMPRESSION_TYPE_CONFIG), is("snappy")); }
@Test(expected = KafkaResponseGetFailedException.class) public void shouldThrowOnNoneRetriableGetTopicConfigError() { expect(adminClient.describeConfigs(anyObject())) .andReturn(topicConfigResponse(new RuntimeException())); replay(adminClient); KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient); final Map<String, String> config = kafkaTopicClient.getTopicConfig("fred"); assertThat(config.get(TopicConfig.RETENTION_MS_CONFIG), is("12345")); assertThat(config.get(TopicConfig.COMPRESSION_TYPE_CONFIG), is("snappy")); }
@Test public void shouldHandleRetriableGetTopicConfigError() { expect(adminClient.describeConfigs(anyObject())) .andReturn(topicConfigResponse(new DisconnectException())) .andReturn(topicConfigResponse( "fred", overriddenConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "12345"), defaultConfigEntry(TopicConfig.COMPRESSION_TYPE_CONFIG, "producer") )); replay(adminClient); KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient); final Map<String, String> config = kafkaTopicClient.getTopicConfig("fred"); assertThat(config.get(TopicConfig.RETENTION_MS_CONFIG), is("12345")); assertThat(config.get(TopicConfig.COMPRESSION_TYPE_CONFIG), is("producer")); }