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

Adding docs for Meter filtering

Browse Source 
KafkaBinderMetrics docs for meter fitlering.
Refact METRIC_NAME to OFFSET_LAG_METRIC_NAME and make it public.
This commit is contained in:
Soby Chacko
2020-12-04 16:45:21 -05:00
committed by Gary Russell
parent 42c9af019e
commit afe13cc045
3 changed files with 26 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
docs/src/main/asciidoc/overview.adoc
View File

@@ -684,9 +684,25 @@ You can consume these exceptions with your own Spring Integration flow.
Kafka binder module exposes the following metrics:
`spring.cloud.stream.binder.kafka.offset`: This metric indicates how many messages have not been yet consumed from a given binder's topic by a given consumer group.
The metrics provided are based on the Mircometer metrics library. The metric contains the consumer group information, topic and the actual lag in committed offset from the latest offset on the topic.
The metrics provided are based on the Micrometer library.
The binder creates the `KafkaBinderMetrics` bean if Micrometer is on the classpath and no other such beans provided by the application.
The metric contains the consumer group information, topic and the actual lag in committed offset from the latest offset on the topic.
This metric is particularly useful for providing auto-scaling feedback to a PaaS platform.
You can exclude `KafkaBinderMetrics` from creating the necessary infrastructure like consumers and then reporting the metrics by providing the following component in the application.
```
@Component
class NoOpBindingMeters {
NoOpBindingMeters(MeterRegistry registry) {
registry.config().meterFilter(
MeterFilter.denyNameStartsWith(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME));
}
}
```
More details on how to suppress meters selectively can be found https://micrometer.io/docs/concepts#_meter_filters[here].
[[kafka-tombstones]]
=== Tombstone Records (null record values)

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

@@ -70,7 +70,7 @@ public class KafkaBinderMetrics
private static final Log LOG = LogFactory.getLog(KafkaBinderMetrics.class);
static final String METRIC_NAME = "spring.cloud.stream.binder.kafka.offset";
public static final String OFFSET_LAG_METRIC_NAME = "spring.cloud.stream.binder.kafka.offset";
private final KafkaMessageChannelBinder binder;
@@ -125,7 +125,7 @@ public class KafkaBinderMetrics
String topic = topicInfo.getKey();
String group = topicInfo.getValue().getConsumerGroup();
final Gauge register = Gauge.builder(METRIC_NAME, this,
final Gauge register = Gauge.builder(OFFSET_LAG_METRIC_NAME, this,
(o) -> computeAndGetUnconsumedMessages(topic, group)).tag("group", group)
.tag("topic", topic)
.description("Unconsumed messages for a particular group and topic")

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

@@ -99,7 +99,7 @@ public class KafkaBinderMetricsTest {
.willReturn(partitions);
metrics.bindTo(meterRegistry);
assertThat(meterRegistry.getMeters()).hasSize(1);
assertThat(meterRegistry.get(KafkaBinderMetrics.METRIC_NAME)
assertThat(meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
.tag("group", "group1-metrics").tag("topic", TEST_TOPIC).gauge().value())
.isEqualTo(500.0);
}
@@ -144,7 +144,7 @@ public class KafkaBinderMetricsTest {
.willReturn(partitions);
metrics.bindTo(meterRegistry);
assertThat(meterRegistry.getMeters()).hasSize(1);
assertThat(meterRegistry.get(KafkaBinderMetrics.METRIC_NAME)
assertThat(meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
.tag("group", "group2-metrics").tag("topic", TEST_TOPIC).gauge().value())
.isEqualTo(1000.0);
}
@@ -158,7 +158,7 @@ public class KafkaBinderMetricsTest {
.willReturn(partitions);
metrics.bindTo(meterRegistry);
assertThat(meterRegistry.getMeters()).hasSize(1);
assertThat(meterRegistry.get(KafkaBinderMetrics.METRIC_NAME)
assertThat(meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
.tag("group", "group3-metrics").tag("topic", TEST_TOPIC).gauge().value())
.isEqualTo(1000.0);
}
@@ -179,7 +179,7 @@ public class KafkaBinderMetricsTest {
metrics.bindTo(meterRegistry);
Gauge gauge = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME)
Gauge gauge = meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
.tag("group", "group4-metrics").tag("topic", TEST_TOPIC).gauge();
gauge.value();
assertThat(gauge.value()).isEqualTo(1000.0);
@@ -201,7 +201,7 @@ public class KafkaBinderMetricsTest {
metrics.bindTo(meterRegistry);
Gauge gauge = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME)
Gauge gauge = meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
.tag("group", "group5-metrics").tag("topic", TEST_TOPIC).gauge();
assertThat(gauge.value()).isEqualTo(0);
assertThat(gauge.value()).isEqualTo(1000.0);
@@ -231,9 +231,9 @@ public class KafkaBinderMetricsTest {
.willReturn(java.util.Collections
.singletonMap(new TopicPartition("test2", 0), 50L));
Gauge gauge1 = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME)
Gauge gauge1 = meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
.tag("group", "group1-metrics").tag("topic", TEST_TOPIC).gauge();
Gauge gauge2 = meterRegistry.get(KafkaBinderMetrics.METRIC_NAME)
Gauge gauge2 = meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
.tag("group", "group2-metrics").tag("topic", "test2").gauge();
gauge1.value();
gauge2.value();