SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

KafkaBinderMetrics NoopGauge/filtering changes (#1001)

Browse Source 
* 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
This commit is contained in:
Soby Chacko
2020-12-04 16:04:09 -05:00
committed by GitHub
parent a1b31e67c4
commit 675c2e4940
2 changed files with 33 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java
View File

@@ -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);
}
}
}

26
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
View File

@@ -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<PartitionInfo> 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<TopicPartition, Long> endOffsets = new HashMap<>();