TopicCreator { public void maybeCreateTopic(String topicName) throws ExecutionException, InterruptedException { Collection<NewTopic> topics = new ArrayList<>(); topics.add(new NewTopic(topicName, 1, (short) 1)); if (topicName.toLowerCase().startsWith("test")) { admin.createTopics(topics); ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName); ConfigEntry compaction = new ConfigEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT); Collection<AlterConfigOp> configOp = new ArrayList<AlterConfigOp>(); configOp.add(new AlterConfigOp(compaction, AlterConfigOp.OpType.SET)); Map<ConfigResource, Collection<AlterConfigOp>> alterConfigs = new HashMap<>(); alterConfigs.put(configResource, configOp); admin.incrementalAlterConfigs(alterConfigs).all().get(); } } TopicCreator(AdminClient admin); void maybeCreateTopic(String topicName);  }
@Test public void testCreateTestTopic() throws ExecutionException, InterruptedException { TopicCreator tc = new TopicCreator(admin); tc.maybeCreateTopic("test.is.a.test.topic"); verify(admin, times(1)).createTopics(any()); }
@Test public void testNotTopic() throws ExecutionException, InterruptedException { TopicCreator tc = new TopicCreator(admin); tc.maybeCreateTopic("not.a.test"); verify(admin, never()).createTopics(any()); }