### Question:
TopicServiceImpl implements TopicService { @Override public void increasePartitions(List<ConfiguredTopic> topics) { CreatePartitionsOptions options = dryRun ? new CreatePartitionsOptions().validateOnly(true) : new CreatePartitionsOptions(); Map<String, NewPartitions> newPartitions = topics .stream() .collect(toMap(ConfiguredTopic::getName, t -> NewPartitions.increaseTo(t.getPartitions()))); try { adminClient.createPartitions(newPartitions, options).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 testIncreasePartitions() { TopicService service = new TopicServiceImpl(adminClient, true); CreatePartitionsResult result = mock(CreatePartitionsResult.class); when(result.all()).thenReturn(KafkaFuture.completedFuture(null)); when(adminClient.createPartitions(any(Map.class), any(CreatePartitionsOptions.class))).thenReturn(result); service.increasePartitions(Collections.singletonList( new ConfiguredTopic("test", 3, (short) 2, Collections.emptyMap()))); ArgumentCaptor<Map> increase = ArgumentCaptor.forClass(Map.class); ArgumentCaptor<CreatePartitionsOptions> options = ArgumentCaptor.forClass(CreatePartitionsOptions.class); verify(adminClient).createPartitions((Map<String, NewPartitions>) increase.capture(), options.capture()); Assert.assertEquals(1, increase.getValue().size()); Assert.assertTrue(increase.getValue().containsKey("test")); Assert.assertEquals(3, ((NewPartitions) increase.getValue().get("test")).totalCount()); Assert.assertTrue(options.getValue().validateOnly()); }