diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index c53566eef..5b7af17c3 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-2022 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. @@ -62,6 +62,7 @@ import org.springframework.util.StringUtils; * @author Aldo Sinanaj * @author Lukasz Kaminski * @author Chukwubuikem Ume-Ugwa + * @author Nico Heller */ @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder") public class KafkaBinderConfigurationProperties { @@ -72,6 +73,8 @@ public class KafkaBinderConfigurationProperties { private final Transaction transaction = new Transaction(); + private final Metrics metrics = new Metrics(); + private final KafkaProperties kafkaProperties; /** @@ -150,6 +153,10 @@ public class KafkaBinderConfigurationProperties { return this.transaction; } + public Metrics getMetrics() { + return metrics; + } + public String getKafkaConnectionString() { // We need to do a check on certificate file locations to see if they are given as classpath resources. // If that is the case, then we will move them to a file system location and use those as the certificate locations. @@ -666,4 +673,34 @@ public class KafkaBinderConfigurationProperties { } + public static class Metrics { + /** + * When set to true, the offset lag metric of each consumer topic is computed whenever the metric is queried (default: true). + * When set to false only the periodically calculated offset lag is used. + * See {@link #offsetLagMetricsInterval} for more information. + */ + private boolean defaultOffsetLagMetricsEnabled = true; + + /** + * The interval in which the offset lag for each consumer topic is computed (default: 60 seconds). + * This value is used whenever {@link #defaultOffsetLagMetricsEnabled} is disabled or its computation is taking too long. + */ + private Duration offsetLagMetricsInterval = Duration.ofSeconds(60); + + public boolean isDefaultOffsetLagMetricsEnabled() { + return defaultOffsetLagMetricsEnabled; + } + + public void setDefaultOffsetLagMetricsEnabled(boolean defaultOffsetLagMetricsEnabled) { + this.defaultOffsetLagMetricsEnabled = defaultOffsetLagMetricsEnabled; + } + + public Duration getOffsetLagMetricsInterval() { + return offsetLagMetricsInterval; + } + + public void setOffsetLagMetricsInterval(Duration offsetLagMetricsInterval) { + this.offsetLagMetricsInterval = offsetLagMetricsInterval; + } + } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java index c3cad944f..dcd2f3c90 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java @@ -30,6 +30,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.ToDoubleFunction; import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.MeterRegistry; @@ -64,14 +65,13 @@ import org.springframework.util.ObjectUtils; * @author Gary Russell * @author Lars Bilger * @author Tomek Szmytka + * @author Nico Heller */ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener, AutoCloseable { private static final int DEFAULT_TIMEOUT = 5; - private static final int DELAY_BETWEEN_TASK_EXECUTION = 60; - private static final Log LOG = LogFactory.getLog(KafkaBinderMetrics.class); /** @@ -91,9 +91,9 @@ public class KafkaBinderMetrics private int timeout = DEFAULT_TIMEOUT; - ScheduledExecutorService scheduler; + private final Map lastUnconsumedMessagesValues = new ConcurrentHashMap<>(); - Map unconsumedMessages = new ConcurrentHashMap<>(); + ScheduledExecutorService scheduler; public KafkaBinderMetrics(KafkaMessageChannelBinder binder, KafkaBinderConfigurationProperties binderConfigurationProperties, @@ -141,58 +141,58 @@ public class KafkaBinderMetrics String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); - 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") - .register(registry); + ToDoubleFunction offsetComputation = computeOffsetComputationFunction(topic, group); + final Gauge register = Gauge.builder(OFFSET_LAG_METRIC_NAME, this, offsetComputation) + .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); + lastUnconsumedMessagesValues.put(topic + "-" + group, 0L); + this.scheduler.scheduleWithFixedDelay( + () -> computeUnconsumedMessages(topic, group), + 1, + binderConfigurationProperties.getMetrics().getOffsetLagMetricsInterval().getSeconds(), + TimeUnit.SECONDS + ); } } } - private Runnable computeUnconsumedMessagesRunnable(String topic, String group, Map> metadataConsumers) { - return () -> { - try { - long lag = findTotalTopicGroupLag(topic, group, this.metadataConsumers); - this.unconsumedMessages.put(topic + "-" + group, lag); - } - catch (Exception ex) { - LOG.debug("Cannot generate metric for topic: " + topic, ex); - } - }; + private ToDoubleFunction computeOffsetComputationFunction(String topic, String group) { + if (this.binderConfigurationProperties.getMetrics().isDefaultOffsetLagMetricsEnabled()) { + return (o) -> computeAndGetUnconsumedMessagesWithTimeout(topic, group); + } + else { + return (o) -> lastUnconsumedMessagesValues.get(topic + "-" + group); + } } - private long computeAndGetUnconsumedMessages(String topic, String group) { - ExecutorService exec = Executors.newCachedThreadPool(); - Future future = exec.submit(() -> { - - long lag = 0; - try { - lag = findTotalTopicGroupLag(topic, group, this.metadataConsumers); - } - catch (Exception ex) { - LOG.debug("Cannot generate metric for topic: " + topic, ex); - } - return lag; - }); + private long computeAndGetUnconsumedMessagesWithTimeout(String topic, String group) { + Future future = scheduler.submit(() -> computeUnconsumedMessages(topic, group)); try { return future.get(this.timeout, TimeUnit.SECONDS); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); - return this.unconsumedMessages.getOrDefault(topic + "-" + group, 0L); + return lastUnconsumedMessagesValues.get(topic + "-" + group); } catch (ExecutionException | TimeoutException ex) { - return this.unconsumedMessages.getOrDefault(topic + "-" + group, 0L); + return lastUnconsumedMessagesValues.get(topic + "-" + group); } - finally { - exec.shutdownNow(); + } + + private long computeUnconsumedMessages(String topic, String group) { + long lag = 0; + try { + lag = findTotalTopicGroupLag(topic, group, this.metadataConsumers); + this.lastUnconsumedMessagesValues.put(topic + "-" + group, lag); } + catch (Exception ex) { + LOG.debug("Cannot generate metric for topic: " + topic, ex); + } + return lag; } private long findTotalTopicGroupLag(String topic, String group, Map> metadataConsumers) { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java index 941580b44..3a15c8659 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.kafka; +import java.time.Duration; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -32,8 +33,10 @@ import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.Node; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; +import org.awaitility.Awaitility; import org.junit.Before; import org.junit.Test; +import org.mockito.Answers; import org.mockito.ArgumentMatchers; import org.mockito.Mock; import org.mockito.Mockito; @@ -52,6 +55,7 @@ import static org.mockito.Mockito.mock; * @author Soby Chacko * @author Lars Bilger * @author Tomek Szmytka + * @author Nico Heller */ public class KafkaBinderMetricsTest { @@ -72,7 +76,7 @@ public class KafkaBinderMetricsTest { private Map topicsInUse = new HashMap<>(); - @Mock + @Mock(answer = Answers.RETURNS_DEEP_STUBS) private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties; @Before @@ -82,6 +86,10 @@ public class KafkaBinderMetricsTest { .createConsumer(ArgumentMatchers.any(), ArgumentMatchers.any())) .willReturn(consumer); org.mockito.BDDMockito.given(binder.getTopicsInUse()).willReturn(topicsInUse); + org.mockito.BDDMockito.given(kafkaBinderConfigurationProperties.getMetrics().isDefaultOffsetLagMetricsEnabled()) + .willReturn(true); + org.mockito.BDDMockito.given(kafkaBinderConfigurationProperties.getMetrics().getOffsetLagMetricsInterval()) + .willReturn(Duration.ofSeconds(60)); metrics = new KafkaBinderMetrics(binder, kafkaBinderConfigurationProperties, consumerFactory, null ); @@ -113,6 +121,39 @@ public class KafkaBinderMetricsTest { .isEqualTo(500.0); } + @Test + public void shouldFallbackToScheduledOffsetLagComputationWhenRealtimeOffsetLagIsDisabled() { + final Map committed = new HashMap<>(); + TopicPartition topicPartition = new TopicPartition(TEST_TOPIC, 0); + committed.put(topicPartition, new OffsetAndMetadata(500)); + org.mockito.BDDMockito + .given(consumer.committed(ArgumentMatchers.anySet())) + .willReturn(committed); + 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); + org.mockito.BDDMockito.given(kafkaBinderConfigurationProperties.getMetrics().isDefaultOffsetLagMetricsEnabled()) + .willReturn(false); + metrics.bindTo(meterRegistry); + assertThat(meterRegistry.getMeters()).hasSize(1); + assertThat(meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME) + .tag("group", "group1-metrics").tag("topic", TEST_TOPIC).gauge().value()) + .isEqualTo(0); + + org.mockito.BDDMockito.given(kafkaBinderConfigurationProperties.getMetrics().getOffsetLagMetricsInterval()) + .willReturn(Duration.ofSeconds(1)); + metrics.bindTo(meterRegistry); + Awaitility.waitAtMost(Duration.ofSeconds(5)).untilAsserted(() -> { + assertThat(meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME) + .tag("group", "group1-metrics").tag("topic", TEST_TOPIC).gauge().value()) + .isEqualTo(500); + }); + } + @Test public void shouldNotContainAnyMetricsWhenUsingNoopGauge() { // Adding NoopGauge for the offset metric. diff --git a/docs/src/main/asciidoc/kafka/kafka_overview.adoc b/docs/src/main/asciidoc/kafka/kafka_overview.adoc index 008bdc098..4a1aad30e 100644 --- a/docs/src/main/asciidoc/kafka/kafka_overview.adoc +++ b/docs/src/main/asciidoc/kafka/kafka_overview.adoc @@ -51,6 +51,7 @@ This section contains the configuration options used by the Apache Kafka binder. For common configuration options and properties pertaining to the binder, see the https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/current/reference/html/spring-cloud-stream.html#binding-properties[binding properties] in core documentation. +[[kafka-binder-properties]] ==== Kafka Binder Properties spring.cloud.stream.kafka.binder.brokers:: @@ -160,6 +161,19 @@ This is also true, if this value is present, but the directory cannot be found o + Default: none. +spring.cloud.stream.kafka.binder.metrics.defaultOffsetLagMetricsEnabled:: +When set to true, the offset lag metric of each consumer topic is computed whenever the metric is accessed. +When set to false only the periodically calculated offset lag is used. ++ +Default: true + +spring.cloud.stream.kafka.binder.metrics.offsetLagMetricsInterval:: +The interval in which the offset lag for each consumer topic is computed. +This value is used whenever `metrics.defaultOffsetLagMetricsEnabled` is disabled or its +computation is taking too long. ++ +Default: 60 seconds + [[kafka-consumer-properties]] ==== Kafka Consumer Properties @@ -744,6 +758,9 @@ The binder creates the `KafkaBinderMetrics` bean if Micrometer is on the classpa 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. +The metric collection behaviour can be configured by setting properties in the `spring.cloud.stream.kafka.binder.metrics` namespace, +refer to the <> for more information. + You can exclude `KafkaBinderMetrics` from creating the necessary infrastructure like consumers and then reporting the metrics by providing the following component in the application. ```