KafkaOffsetMetrics { @Scheduled(fixedDelayString = "${refreshPeriod:PT10S}") public void refresh() { Flux .fromIterable(asJavaCollection(adminClient.listAllGroupsFlattened())) .map(GroupOverview::groupId) .doOnNext(groupId -> log.info("Found groupId: {}", groupId)) .flatMap(groupId -> Mono .just(mapAsJavaMap(adminClient.listGroupOffsets(groupId))) .flatMapIterable(Map::entrySet) .map(entry -> toRow(groupId, entry))) .collectList() .doOnNext(rows -> gauge.register(rows, true)) .then() .block(); } KafkaOffsetMetrics(AdminClient adminClient, MeterRegistry registry); @Scheduled(fixedDelayString = "${refreshPeriod:PT10S}") void refresh();  }
@SuppressWarnings("unchecked") @Test public void happyPath() { scala.collection.immutable.List<GroupOverview> listAllGroupsFlattened = asScalaSet( singleton(GroupOverview.apply("groupId", ""))).toList(); when(adminClient.listAllGroupsFlattened()).thenReturn(listAllGroupsFlattened); scala.collection.immutable.Map<TopicPartition, Object> offsets = asScalaMap( singletonMap(new TopicPartition("topicName", 0), (Object) 1L)); when(adminClient.listGroupOffsets("groupId")).thenReturn(offsets); underTest.refresh(); Matcher<Iterable<? extends Tag>> tags = containsInAnyOrder(tag("group", "groupId"), tag("topic", "topicName"), tag("partition", "0")); assertMeter(registry.get("kafka_offset").meter(), tags, 1); }