### Question:
KafkaTopicClientImpl implements KafkaTopicClient { @Override public void createTopic( final String topic, final int numPartitions, final short replicationFactor ) { createTopic(topic, numPartitions, replicationFactor, Collections.emptyMap()); } 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 shouldCreateTopic() { expect(adminClient.listTopics()).andReturn(getListTopicsResult()); expect(adminClient.createTopics(anyObject())).andReturn(getCreateTopicsResult()); replay(adminClient); KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient); kafkaTopicClient.createTopic("test", 1, (short) 1); verify(adminClient); }
@Test public void shouldSetTopicCleanupPolicyToCompact() throws InterruptedException, ExecutionException { expect(adminClient.listTopics()).andReturn(getEmptyListTopicResult()); NewTopic newTopic = new NewTopic(topicName1, 1, (short) 1); newTopic.configs(Collections.singletonMap("cleanup.policy", "compact")); expect(adminClient.createTopics(singleNewTopic(newTopic))).andReturn(getCreateTopicsResult()); replay(adminClient); KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient); kafkaTopicClient.createTopic(topicName1, 1, (short) 1, Collections.singletonMap("cleanup.policy", "compact")); verify(adminClient); }