From 227d8f39f61de95165a48b0fa6e93f3a00ef2ffe Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 27 Feb 2018 18:49:20 -0500 Subject: [PATCH] GH-318: Fix KafkaBinderMetrics for Micrometer Fixes spring-cloud/spring-cloud-stream-binder-kafka#318 * Use `ToDoubleFunction`-based `MeterRegistry.gauge()` variant to really calculate a `lag` at runtime instead of statically defined before * Add `KafkaBinderActuatorTests` integration test to demonstrate how `Binder` is declared in the separate child context and how `KafkaBinderMetrics` is not visible from the parent context. This test also verify the real `gauge` value for the `consumer lag` and should be used in the future to verify the `KafkaBinderMetrics` exposure removing the code after TODO in the `KafkaMetricsTestConfig` --- .../binder/kafka/KafkaBinderMetrics.java | 83 +++++++----- .../integration/KafkaBinderActuatorTests.java | 128 ++++++++++++++++++ 2 files changed, 176 insertions(+), 35 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java 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 b6abaeb28..5acf118a6 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-2017 the original author or authors. + * Copyright 2016-2018 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.MeterRegistry; import io.micrometer.core.instrument.binder.MeterBinder; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.Consumer; @@ -41,6 +42,8 @@ import org.springframework.util.ObjectUtils; * Metrics for Kafka binder. * * @author Henryk Konsek + * @author Soby Chacko + * @author Artem Bilan */ public class KafkaBinderMetrics implements MeterBinder { @@ -57,6 +60,7 @@ public class KafkaBinderMetrics implements MeterBinder { public KafkaBinderMetrics(KafkaMessageChannelBinder binder, KafkaBinderConfigurationProperties binderConfigurationProperties, ConsumerFactory defaultConsumerFactory) { + this.binder = binder; this.binderConfigurationProperties = binderConfigurationProperties; this.defaultConsumerFactory = defaultConsumerFactory; @@ -64,6 +68,7 @@ public class KafkaBinderMetrics implements MeterBinder { public KafkaBinderMetrics(KafkaMessageChannelBinder binder, KafkaBinderConfigurationProperties binderConfigurationProperties) { + this(binder, binderConfigurationProperties, null); } @@ -71,6 +76,7 @@ public class KafkaBinderMetrics implements MeterBinder { public void bindTo(MeterRegistry registry) { for (Map.Entry topicInfo : this.binder.getTopicsInUse() .entrySet()) { + if (!topicInfo.getValue().isConsumerTopic()) { continue; } @@ -78,47 +84,54 @@ public class KafkaBinderMetrics implements MeterBinder { String topic = topicInfo.getKey(); String group = topicInfo.getValue().getConsumerGroup(); - try (Consumer metadataConsumer = createConsumerFactory(group).createConsumer()) { - List partitionInfos = metadataConsumer.partitionsFor(topic); - List topicPartitions = new LinkedList<>(); - for (PartitionInfo partitionInfo : partitionInfos) { - topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition())); - } - Map endOffsets = metadataConsumer.endOffsets(topicPartitions); - long lag = 0; - for (Map.Entry endOffset : endOffsets.entrySet()) { - OffsetAndMetadata current = metadataConsumer.committed(endOffset.getKey()); - if (current != null) { - lag += endOffset.getValue() - current.offset(); - } - else { - lag += endOffset.getValue(); - } - } - registry.gauge(String.format("%s.%s.%s.lag", METRIC_PREFIX, group, topic), lag); + registry.gauge(String.format("%s.%s.%s.lag", METRIC_PREFIX, group, topic), this, + o -> calculateConsumerLagOnTopic(topic, group)); + } + } + + private double calculateConsumerLagOnTopic(String topic, String group) { + long lag = 0; + try (Consumer metadataConsumer = createConsumerFactory(group).createConsumer()) { + List partitionInfos = metadataConsumer.partitionsFor(topic); + List topicPartitions = new LinkedList<>(); + for (PartitionInfo partitionInfo : partitionInfos) { + topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition())); } - catch (Exception e) { - LOG.debug("Cannot generate metric for topic: " + topic, e); + Map endOffsets = metadataConsumer.endOffsets(topicPartitions); + + for (Map.Entry endOffset : endOffsets.entrySet()) { + OffsetAndMetadata current = metadataConsumer.committed(endOffset.getKey()); + if (current != null) { + lag += endOffset.getValue() - current.offset(); + } + else { + lag += endOffset.getValue(); + } } } + catch (Exception e) { + LOG.debug("Cannot generate metric for topic: " + topic, e); + } + return lag; } private ConsumerFactory createConsumerFactory(String group) { - if (defaultConsumerFactory != null) { - return defaultConsumerFactory; + if (this.defaultConsumerFactory == null) { + Map props = new HashMap<>(); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConsumerConfiguration())) { + props.putAll(binderConfigurationProperties.getConsumerConfiguration()); + } + if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) { + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + this.binderConfigurationProperties.getKafkaConnectionString()); + } + props.put("group.id", group); + this.defaultConsumerFactory = new DefaultKafkaConsumerFactory<>(props); } - Map props = new HashMap<>(); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); - props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); - if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConsumerConfiguration())) { - props.putAll(binderConfigurationProperties.getConsumerConfiguration()); - } - if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) { - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, - this.binderConfigurationProperties.getKafkaConnectionString()); - } - props.put("group.id", group); - return new DefaultKafkaConsumerFactory<>(props); + + return this.defaultConsumerFactory; } } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java new file mode 100644 index 000000000..843f2c2a4 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java @@ -0,0 +1,128 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.integration; + +import java.util.ArrayList; +import java.util.List; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.binder.MeterBinder; +import io.micrometer.core.instrument.search.Search; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.DefaultBinderFactory; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.context.ApplicationListener; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Artem Bilan + * + * @since 2.0 + */ +@RunWith(SpringRunner.class) +@SpringBootTest( + webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = "spring.cloud.stream.bindings.input.group=" + KafkaBinderActuatorTests.TEST_CONSUMER_GROUP) +public class KafkaBinderActuatorTests { + + static final String TEST_CONSUMER_GROUP = "testGroup"; + + private static final String KAFKA_BROKERS_PROPERTY = "spring.kafka.bootstrap-servers"; + + @ClassRule + public static KafkaEmbedded kafkaEmbedded = new KafkaEmbedded(1, true); + + @BeforeClass + public static void setup() { + System.setProperty(KAFKA_BROKERS_PROPERTY, kafkaEmbedded.getBrokersAsString()); + } + + @AfterClass + public static void clean() { + System.clearProperty(KAFKA_BROKERS_PROPERTY); + } + + @Autowired + private MeterRegistry meterRegistry; + + @Autowired + private KafkaTemplate kafkaTemplate; + + @Test + public void testKafkaBinderMetricsExposed() { + Search search = this.meterRegistry.find( + String.format("%s.%s.%s.lag", "spring.cloud.stream.binder.kafka", TEST_CONSUMER_GROUP, Sink.INPUT)); + + assertThat(search.gauge()).isNotNull(); + + this.kafkaTemplate.send(Sink.INPUT, null, "foo".getBytes()); + this.kafkaTemplate.flush(); + + assertThat(search.gauge().value()).isGreaterThan(0); + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + public static class KafkaMetricsTestConfig implements ApplicationListener { + + @StreamListener(Sink.INPUT) + public void process(String payload) throws InterruptedException { + // Artificial slow listener to emulate consumer lag + Thread.sleep(1000); + } + + + // TODO until the fix on the core level. See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/250 + @Autowired + private MeterRegistry meterRegistry; + + private final List springCloudStreamMeterBinders = new ArrayList<>(); + + @Bean + public DefaultBinderFactory.Listener listenerForMeterBinders() { + + return (configurationName, binderContext) -> { + springCloudStreamMeterBinders.add(binderContext.getBean(MeterBinder.class)); + }; + + } + + @Override + public void onApplicationEvent(ApplicationReadyEvent event) { + this.springCloudStreamMeterBinders.forEach(b -> b.bindTo(this.meterRegistry)); + } + + } + +}