### Question:
TopicServiceImpl implements TopicService { @Override public Map<String, ConfiguredTopic> listExisting(boolean excludeInternal) { try { Set<String> topics = adminClient .listTopics(excludeInternal ? EXCLUDE_INTERNAL : INCLUDE_INTERNAL) .names().get(); Collection<TopicDescription> topicDescriptions = adminClient.describeTopics(topics).all().get().values(); List<ConfigResource> resources = topics .stream() .map(t -> new ConfigResource(Type.TOPIC, t)) .collect(toList()); Map<ConfigResource, KafkaFuture<Config>> topicConfigs = adminClient.describeConfigs(resources).values(); return topicDescriptions .stream() .map(td -> configuredTopic(td, topicConfigs.get(new ConfigResource(Type.TOPIC, td.name())))) .filter(t -> !INTERNAL_TOPIC.test(t)) .collect(toMap(ConfiguredTopic::getName, td -> td)); } 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 testListExisting() { Cluster cluster = createCluster(1); TopicPartitionInfo tp = new TopicPartitionInfo(0, cluster.nodeById(0), cluster.nodes(), Collections.emptyList()); ConfigEntry configEntry = new ConfigEntry("k", "v"); KafkaFuture<Config> kfc = KafkaFuture.completedFuture(new Config(Collections.singletonList(configEntry))); Set<String> topicNames = new HashSet<>(Arrays.asList("a", "b", "_c")); Map<String, TopicDescription> tds = new HashMap<String, TopicDescription>() { { put("a", new TopicDescription("a", false, Collections.singletonList(tp))); put("b", new TopicDescription("b", false, Collections.singletonList(tp))); put("c", new TopicDescription("_c", false, Collections.singletonList(tp))); } }; Map<ConfigResource, KafkaFuture<Config>> configs = new HashMap<ConfigResource, KafkaFuture<Config>>() { { put(new ConfigResource(TOPIC, "a"), kfc); put(new ConfigResource(TOPIC, "b"), kfc); put(new ConfigResource(TOPIC, "_c"), kfc); } }; TopicService service = new TopicServiceImpl(adminClient, true); ListTopicsResult listTopicsResult = mock(ListTopicsResult.class); DescribeTopicsResult describeTopicsResult = mock(DescribeTopicsResult.class); DescribeConfigsResult describeConfigsResult = mock(DescribeConfigsResult.class); when(describeTopicsResult.all()).thenReturn(KafkaFuture.completedFuture(tds)); when(listTopicsResult.names()).thenReturn(KafkaFuture.completedFuture(topicNames)); when(describeConfigsResult.values()).thenReturn(configs); when(adminClient.listTopics(any(ListTopicsOptions.class))).thenReturn(listTopicsResult); when(adminClient.describeTopics(topicNames)).thenReturn(describeTopicsResult); when(adminClient.describeConfigs(any(Collection.class))).thenReturn(describeConfigsResult); Map<String, ConfiguredTopic> actual = service.listExisting(true); Assert.assertEquals(2, actual.size()); Assert.assertEquals(new HashSet<>(Arrays.asList("a", "b")), actual.keySet()); }