diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 242d174f8..679128984 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -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) 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 ad43c89d6..bc212b331 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 @@ -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") 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 071339f57..480d7795c 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 @@ -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();