From 675c2e494088ea761f13f6509336798c5655a630 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 4 Dec 2020 16:04:09 -0500 Subject: [PATCH] KafkaBinderMetrics NoopGauge/filtering changes (#1001) * KafkaBinderMetrics NoopGauge/filtering changes Fixing the problem of KafkBinderMetrics scheduled task for finding the offset lag gets triggered, even when the guage is registered as a NoopGague in the MeterRegistry. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/995 * Addressing PR review comments * Fix typo --- .../binder/kafka/KafkaBinderMetrics.java | 15 ++++++----- .../binder/kafka/KafkaBinderMetricsTest.java | 26 +++++++++++++++++-- 2 files changed, 33 insertions(+), 8 deletions(-) diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java index 6449be335..ad43c89d6 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -32,6 +32,7 @@ import java.util.concurrent.TimeoutException; import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.binder.MeterBinder; +import io.micrometer.core.instrument.noop.NoopGauge; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.Consumer; @@ -124,15 +125,17 @@ public class KafkaBinderMetrics String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); - //Schedule a task to compute the unconsumed messages for this group/topic every minute. - this.scheduler.scheduleWithFixedDelay(computeUnconsumedMessagesRunnable(topic, group, this.metadataConsumers), - 10, DELAY_BETWEEN_TASK_EXECUTION, TimeUnit.SECONDS); - - Gauge.builder(METRIC_NAME, this, + final Gauge register = Gauge.builder(METRIC_NAME, this, (o) -> computeAndGetUnconsumedMessages(topic, group)).tag("group", group) .tag("topic", topic) .description("Unconsumed messages for a particular group and topic") .register(registry); + + if (!(register instanceof NoopGauge)) { + //Schedule a task to compute the unconsumed messages for this group/topic every minute. + this.scheduler.scheduleWithFixedDelay(computeUnconsumedMessagesRunnable(topic, group, this.metadataConsumers), + 10, DELAY_BETWEEN_TASK_EXECUTION, TimeUnit.SECONDS); + } } } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java index 7913a95b1..071339f57 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,6 +23,7 @@ import java.util.Map; import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.config.MeterFilter; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.OffsetAndMetadata; @@ -73,7 +74,7 @@ public class KafkaBinderMetricsTest { @Before public void setup() { - MockitoAnnotations.initMocks(this); + MockitoAnnotations.openMocks(this); org.mockito.BDDMockito.given(consumerFactory .createConsumer(ArgumentMatchers.any(), ArgumentMatchers.any())) .willReturn(consumer); @@ -103,6 +104,27 @@ public class KafkaBinderMetricsTest { .isEqualTo(500.0); } + @Test + public void shouldNotContainAnyMetricsWhenUsingNoopGauge() { + // Adding NoopGauge for the offset metric. + meterRegistry.config().meterFilter( + MeterFilter.denyNameStartsWith("spring.cloud.stream.binder.kafka.offset")); + + // Because we have NoopGauge for the offset metric in the meter registry, none of these expectations matter. + org.mockito.BDDMockito + .given(consumer.committed(ArgumentMatchers.any(TopicPartition.class))) + .willReturn(new OffsetAndMetadata(500)); + List partitions = partitions(new Node(0, null, 0)); + topicsInUse.put(TEST_TOPIC, + new TopicInformation("group1-metrics", partitions, false)); + org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)) + .willReturn(partitions); + metrics.bindTo(meterRegistry); + + // Because of the NoopGauge, the meterRegistry should contain no metric. + assertThat(meterRegistry.getMeters()).hasSize(0); + } + @Test public void shouldSumUpPartitionsLags() { Map endOffsets = new HashMap<>();