RetentionByTopicFunction { public Map<String, Duration> apply(Collection<String> topics) { return Flux .fromIterable(topics) .map(name -> new ConfigResource(TOPIC, name)) .collectList() .map(crs -> client.describeConfigs(crs).all()) .map(KafkaFutures::join) .flatMapIterable(Map::entrySet) .collectMap(e -> e.getKey().name(), this::retention) .block(); }  Map<String, Duration> apply(Collection<String> topics);  }
@Test public void typical() throws Exception { String topic = "topic"; Collection<String> topics = singleton(topic); ConfigResource configResource = new ConfigResource(TOPIC, topic); Config config = new Config(singleton(new ConfigEntry("retention.ms", "1"))); KafkaFuture<Map<ConfigResource, Config>> kafkaFuture = completedFuture(singletonMap(configResource, config)); doReturn(describeConfigsResult).when(adminClient).describeConfigs(any()); doReturn(kafkaFuture).when(describeConfigsResult).all(); Map<String, Duration> result = underTest.apply(topics); assertThat(result.size(), is(1)); Duration retention = result.get(topic); assertThat(retention, is(Duration.ofMillis(1))); }