### Question:
TopicServiceImpl implements TopicService { @Override public void create(List<ConfiguredTopic> topics) { try { CreateTopicsOptions options = dryRun ? new CreateTopicsOptions().validateOnly(true) : new CreateTopicsOptions(); List<NewTopic> newTopics = topics .stream() .map(t -> new NewTopic(t.getName(), t.getPartitions(), t.getReplicationFactor()).configs(t.getConfig())) .collect(toList()); CreateTopicsResult result = adminClient.createTopics(newTopics, options); result.all().get(); } catch (InterruptedException | ExecutionException e) { throw new RuntimeException(e); } } TopicServiceImpl(AdminClient adminClient, boolean dryRun); @Override Map<String, ConfiguredTopic> listExisting(boolean excludeInternal); @Override void create(List<ConfiguredTopic> topics); @Override void increasePartitions(List<ConfiguredTopic> topics); @Override void alterConfiguration(List<ConfiguredTopic> topics); @Override void delete(List<ConfiguredTopic> topics);  }### Answer:
@Test public void testCreate() { TopicService service = new TopicServiceImpl(adminClient, true); CreateTopicsResult createTopicsResult = mock(CreateTopicsResult.class); when(createTopicsResult.all()).thenReturn(KafkaFuture.completedFuture(null)); when(adminClient.createTopics(any(Collection.class), any(CreateTopicsOptions.class))).thenReturn(createTopicsResult); service.create(Collections.singletonList( new ConfiguredTopic("test", 1, (short) 2, Collections.emptyMap()))); ArgumentCaptor<List> newTopics = ArgumentCaptor.forClass(List.class); ArgumentCaptor<CreateTopicsOptions> options = ArgumentCaptor.forClass(CreateTopicsOptions.class); verify(adminClient).createTopics((Collection<NewTopic>) newTopics.capture(), options.capture()); Assert.assertEquals(1, newTopics.getValue().size()); Assert.assertEquals("test", ((NewTopic) newTopics.getValue().get(0)).name()); Assert.assertEquals(2, ((NewTopic) newTopics.getValue().get(0)).replicationFactor()); Assert.assertTrue(options.getValue().shouldValidateOnly()); }