### Question:
TopicServiceImpl implements TopicService { static ConfiguredTopic configuredTopic(TopicDescription td, KafkaFuture<Config> ktc) { int partitions = td.partitions().size(); short replication = (short) td.partitions().iterator().next().replicas().size(); try { Config tc = ktc.get(); Map<String, String> configMap = tc .entries() .stream() .filter(TopicServiceImpl::isNonDefault) .collect(toMap(ConfigEntry::name, ConfigEntry::value)); return new ConfiguredTopic(td.name(), partitions, replication, configMap); } 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 testConfiguredTopic() { Cluster cluster = createCluster(1); TopicPartitionInfo tp = new TopicPartitionInfo(0, cluster.nodeById(0), cluster.nodes(), Collections.emptyList()); TopicDescription td = new TopicDescription("test", false, Collections.singletonList(tp)); ConfigEntry configEntry = mock(ConfigEntry.class); when(configEntry.source()).thenReturn(ConfigEntry.ConfigSource.DYNAMIC_DEFAULT_BROKER_CONFIG); KafkaFuture<Config> kfc = KafkaFuture.completedFuture(new Config(Collections.singletonList(configEntry))); ConfiguredTopic expected = new ConfiguredTopic("test", 1, (short) 1, Collections.emptyMap()); Assert.assertEquals(expected, TopicServiceImpl.configuredTopic(td, kfc)); }