From 9ad04882c8c36e812432a1c3c2a5148a7821dc6e Mon Sep 17 00:00:00 2001 From: Henryk Konsek Date: Mon, 22 May 2017 15:47:01 +0200 Subject: [PATCH] Added Kafka binder lag metrics. Fix #152 Metrics should be divided by group ID of the binding. TopicInformation should carry optional group of the consumer. Improved tests coverage. Added lag metric documentation. --- .../src/main/asciidoc/index.adoc | 3 +- .../src/main/asciidoc/metrics.adoc | 10 ++ .../kafka/KafkaBinderHealthIndicator.java | 8 +- .../binder/kafka/KafkaBinderMetrics.java | 126 ++++++++++++++++ .../kafka/KafkaMessageChannelBinder.java | 110 +++++++++----- .../config/KafkaBinderConfiguration.java | 37 +++-- ...BinderAutoConfigurationPropertiesTest.java | 36 +++-- ...afkaBinderConfigurationPropertiesTest.java | 33 +++-- .../kafka/KafkaBinderConfigurationTest.java | 3 +- .../kafka/KafkaBinderHealthIndicatorTest.java | 16 +- .../binder/kafka/KafkaBinderMetricsTest.java | 137 ++++++++++++++++++ 11 files changed, 437 insertions(+), 82 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/metrics.adoc create mode 100644 spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java create mode 100644 spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/index.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/index.adoc index e48d31838..a58d4b502 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/index.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/index.adoc @@ -1,6 +1,6 @@ [[spring-cloud-stream-binder-kafka-reference]] = Spring Cloud Stream Kafka Binder Reference Guide -Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein +Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Henryk Konsek :doctype: book :toc: :toclevels: 4 @@ -24,6 +24,7 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat = Reference Guide include::overview.adoc[] include::dlq.adoc[] +include::metrics.adoc[] = Appendices [appendix] diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/metrics.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/metrics.adoc new file mode 100644 index 000000000..1a5a9d952 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/metrics.adoc @@ -0,0 +1,10 @@ +[[kafka-metrics]] +== Kafka metrics + +Kafka binder module exposes the following metrics: + +`spring.cloud.stream.binder.kafka.someGroup.someTopic.lag` - this metric indicates how many messages +have not been yet consumed from given binder's topic (`someTopic`) by given consumer group (`someGroup`). +For example if the value of the metric `spring.cloud.stream.binder.kafka.myGroup.myTopic.lag` is `1000`, then +consumer group `myGroup` has `1000` messages to waiting to be consumed from topic `myTopic`. This metric is +particularly useful to provide auto-scaling feedback to PaaS platform of your choice. \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java index 5c7f89ebb..99e0a9e44 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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 org.springframework.kafka.core.ConsumerFactory; * * @author Ilayaperumal Gopinathan * @author Marius Bogoevici + * @author Henryk Konsek */ public class KafkaBinderHealthIndicator implements HealthIndicator { @@ -53,8 +54,9 @@ public class KafkaBinderHealthIndicator implements HealthIndicator { for (String topic : this.binder.getTopicsInUse().keySet()) { List partitionInfos = metadataConsumer.partitionsFor(topic); for (PartitionInfo partitionInfo : partitionInfos) { - if (this.binder.getTopicsInUse().get(topic).contains(partitionInfo) && partitionInfo.leader() - .id() == -1) { + if (this.binder.getTopicsInUse().get(topic).getPartitionInfos().contains(partitionInfo) + && partitionInfo.leader() + .id() == -1) { downMessages.add(partitionInfo.toString()); } } 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 new file mode 100644 index 000000000..d3d5c7c82 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java @@ -0,0 +1,126 @@ +/* + * Copyright 2017 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; + +import java.util.Collection; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.boot.actuate.endpoint.PublicMetrics; +import org.springframework.boot.actuate.metrics.Metric; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.util.ObjectUtils; + +/** + * Metrics for Kafka binder. + * + * @author Henryk Konsek + */ +public class KafkaBinderMetrics implements PublicMetrics { + + private final static Logger LOG = LoggerFactory.getLogger(KafkaBinderMetrics.class); + + static final String METRIC_PREFIX = "spring.cloud.stream.binder.kafka"; + + private final KafkaMessageChannelBinder binder; + + private final KafkaBinderConfigurationProperties binderConfigurationProperties; + + private ConsumerFactory defaultConsumerFactory; + + public KafkaBinderMetrics(KafkaMessageChannelBinder binder, + KafkaBinderConfigurationProperties binderConfigurationProperties, + ConsumerFactory defaultConsumerFactory) { + this.binder = binder; + this.binderConfigurationProperties = binderConfigurationProperties; + this.defaultConsumerFactory = defaultConsumerFactory; + } + + public KafkaBinderMetrics(KafkaMessageChannelBinder binder, + KafkaBinderConfigurationProperties binderConfigurationProperties) { + this(binder, binderConfigurationProperties, null); + } + + @Override + public Collection> metrics() { + List> metrics = new LinkedList<>(); + for (Map.Entry topicInfo : this.binder.getTopicsInUse() + .entrySet()) { + if (!topicInfo.getValue().isConsumerTopic()) { + continue; + } + + 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(); + } + } + metrics.add(new Metric<>(String.format("%s.%s.%s.lag", METRIC_PREFIX, group, topic), lag)); + } + catch (Exception e) { + LOG.debug("Cannot generate metric for topic: " + topic, e); + } + } + return metrics; + } + + private ConsumerFactory createConsumerFactory(String group) { + if (defaultConsumerFactory != null) { + return defaultConsumerFactory; + } + 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.getConfiguration())) { + props.putAll(binderConfigurationProperties.getConfiguration()); + } + 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); + } + +} \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index a22959e1c..0caf07795 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -87,8 +87,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * @author Doug Saus */ public class KafkaMessageChannelBinder extends - AbstractMessageChannelBinder, - ExtendedProducerProperties, KafkaTopicProvisioner> + AbstractMessageChannelBinder, ExtendedProducerProperties, KafkaTopicProvisioner> implements ExtendedPropertiesBinder { private final KafkaBinderConfigurationProperties configurationProperties; @@ -97,7 +96,7 @@ public class KafkaMessageChannelBinder extends private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); - private final Map> topicsInUse = new HashMap<>(); + private final Map topicsInUse = new HashMap<>(); public KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties, KafkaTopicProvisioner provisioningProvider) { @@ -129,7 +128,7 @@ public class KafkaMessageChannelBinder extends this.producerListener = producerListener; } - Map> getTopicsInUse() { + Map getTopicsInUse() { return this.topicsInUse; } @@ -147,7 +146,8 @@ public class KafkaMessageChannelBinder extends protected MessageHandler createProducerMessageHandler(final ProducerDestination destination, ExtendedProducerProperties producerProperties) throws Exception { final DefaultKafkaProducerFactory producerFB = getProducerFactory(producerProperties); - Collection partitions = provisioningProvider.getPartitionsForTopic(producerProperties.getPartitionCount(), + Collection partitions = provisioningProvider.getPartitionsForTopic( + producerProperties.getPartitionCount(), false, new Callable>() { @Override @@ -155,7 +155,7 @@ public class KafkaMessageChannelBinder extends return producerFB.createProducer().partitionsFor(destination.getName()); } }); - this.topicsInUse.put(destination.getName(), partitions); + this.topicsInUse.put(destination.getName(), new TopicInformation(null, partitions)); if (producerProperties.getPartitionCount() < partitions.size()) { if (this.logger.isInfoEnabled()) { this.logger.info("The `partitionCount` of the producer for topic " + destination.getName() + " is " @@ -168,7 +168,8 @@ public class KafkaMessageChannelBinder extends if (this.producerListener != null) { kafkaTemplate.setProducerListener(this.producerListener); } - return new ProducerConfigurationMessageHandler(kafkaTemplate, destination.getName(), producerProperties, producerFB); + return new ProducerConfigurationMessageHandler(kafkaTemplate, destination.getName(), producerProperties, + producerFB); } private DefaultKafkaProducerFactory getProducerFactory( @@ -186,10 +187,12 @@ public class KafkaMessageChannelBinder extends props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); } if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { - props.put(ProducerConfig.BATCH_SIZE_CONFIG, String.valueOf(producerProperties.getExtension().getBufferSize())); + props.put(ProducerConfig.BATCH_SIZE_CONFIG, + String.valueOf(producerProperties.getExtension().getBufferSize())); } if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { - props.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(producerProperties.getExtension().getBatchTimeout())); + props.put(ProducerConfig.LINGER_MS_CONFIG, + String.valueOf(producerProperties.getExtension().getBatchTimeout())); } if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, @@ -210,8 +213,10 @@ public class KafkaMessageChannelBinder extends Assert.isTrue(!anonymous || !extendedConsumerProperties.getExtension().isEnableDlq(), "DLQ support is not available for anonymous subscriptions"); String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; - final ConsumerFactory consumerFactory = createKafkaConsumerFactory(anonymous, consumerGroup, extendedConsumerProperties); - int partitionCount = extendedConsumerProperties.getInstanceCount() * extendedConsumerProperties.getConcurrency(); + final ConsumerFactory consumerFactory = createKafkaConsumerFactory(anonymous, consumerGroup, + extendedConsumerProperties); + int partitionCount = extendedConsumerProperties.getInstanceCount() + * extendedConsumerProperties.getConcurrency(); Collection allPartitions = provisioningProvider.getPartitionsForTopic(partitionCount, extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(), @@ -232,33 +237,37 @@ public class KafkaMessageChannelBinder extends listenedPartitions = new ArrayList<>(); for (PartitionInfo partition : allPartitions) { // divide partitions across modules - if ((partition.partition() % extendedConsumerProperties.getInstanceCount()) == extendedConsumerProperties.getInstanceIndex()) { + if ((partition.partition() + % extendedConsumerProperties.getInstanceCount()) == extendedConsumerProperties + .getInstanceIndex()) { listenedPartitions.add(partition); } } } - this.topicsInUse.put(destination.getName(), listenedPartitions); + this.topicsInUse.put(destination.getName(), new TopicInformation(group, listenedPartitions)); Assert.isTrue(!CollectionUtils.isEmpty(listenedPartitions), "A list of partitions must be provided"); final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets( listenedPartitions); - final ContainerProperties containerProperties = - anonymous || extendedConsumerProperties.getExtension().isAutoRebalanceEnabled() ? - new ContainerProperties(destination.getName()) : new ContainerProperties(topicPartitionInitialOffsets); + final ContainerProperties containerProperties = anonymous + || extendedConsumerProperties.getExtension().isAutoRebalanceEnabled() + ? new ContainerProperties(destination.getName()) + : new ContainerProperties(topicPartitionInitialOffsets); int concurrency = Math.min(extendedConsumerProperties.getConcurrency(), listenedPartitions.size()); - final ConcurrentMessageListenerContainer messageListenerContainer = - new ConcurrentMessageListenerContainer( - consumerFactory, containerProperties) { + final ConcurrentMessageListenerContainer messageListenerContainer = new ConcurrentMessageListenerContainer( + consumerFactory, containerProperties) { - @Override - public void stop(Runnable callback) { - super.stop(callback); - } - }; + @Override + public void stop(Runnable callback) { + super.stop(callback); + } + }; messageListenerContainer.setConcurrency(concurrency); - messageListenerContainer.getContainerProperties().setAckOnError(isAutoCommitOnError(extendedConsumerProperties)); + messageListenerContainer.getContainerProperties() + .setAckOnError(isAutoCommitOnError(extendedConsumerProperties)); if (!extendedConsumerProperties.getExtension().isAutoCommitOffset()) { - messageListenerContainer.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL); + messageListenerContainer.getContainerProperties() + .setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL); } if (this.logger.isDebugEnabled()) { this.logger.debug( @@ -268,14 +277,14 @@ public class KafkaMessageChannelBinder extends this.logger.debug( "Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions)); } - final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = - new KafkaMessageDrivenChannelAdapter<>( - messageListenerContainer); + final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter<>( + messageListenerContainer); kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory()); final RetryTemplate retryTemplate = buildRetryTemplate(extendedConsumerProperties); kafkaMessageDrivenChannelAdapter.setRetryTemplate(retryTemplate); if (extendedConsumerProperties.getExtension().isEnableDlq()) { - DefaultKafkaProducerFactory producerFactory = getProducerFactory(new ExtendedProducerProperties<>(new KafkaProducerProperties())); + DefaultKafkaProducerFactory producerFactory = getProducerFactory( + new ExtendedProducerProperties<>(new KafkaProducerProperties())); final KafkaTemplate kafkaTemplate = new KafkaTemplate<>(producerFactory); messageListenerContainer.getContainerProperties().setErrorHandler(new ErrorHandler() { @@ -285,9 +294,11 @@ public class KafkaMessageChannelBinder extends : null; final byte[] payload = message.value() != null ? Utils.toArray(ByteBuffer.wrap((byte[]) message.value())) : null; - String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName()) ? - extendedConsumerProperties.getExtension().getDlqName() : "error." + destination.getName() + "." + group; - ListenableFuture> sentDlq = kafkaTemplate.send(dlqName, message.partition(), key, payload); + String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName()) + ? extendedConsumerProperties.getExtension().getDlqName() + : "error." + destination.getName() + "." + group; + ListenableFuture> sentDlq = kafkaTemplate.send(dlqName, + message.partition(), key, payload); sentDlq.addCallback(new ListenableFutureCallback>() { StringBuilder sb = new StringBuilder().append(" a message with key='") .append(toDisplayString(ObjectUtils.nullSafeToString(key), 50)).append("'") @@ -337,7 +348,8 @@ public class KafkaMessageChannelBinder extends props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) { - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, consumerProperties.getExtension().getStartOffset().name()); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, + consumerProperties.getExtension().getStartOffset().name()); } return new DefaultKafkaConsumerFactory<>(props); @@ -351,8 +363,8 @@ public class KafkaMessageChannelBinder extends private TopicPartitionInitialOffset[] getTopicPartitionInitialOffsets( Collection listenedPartitions) { - final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = - new TopicPartitionInitialOffset[listenedPartitions.size()]; + final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = new TopicPartitionInitialOffset[listenedPartitions + .size()]; int i = 0; for (PartitionInfo partition : listenedPartitions) { @@ -415,4 +427,30 @@ public class KafkaMessageChannelBinder extends return this.running; } } + + public static class TopicInformation { + + private final String consumerGroup; + + private final Collection partitionInfos; + + public TopicInformation(String consumerGroup, Collection partitionInfos) { + this.consumerGroup = consumerGroup; + this.partitionInfos = partitionInfos; + } + + public String getConsumerGroup() { + return consumerGroup; + } + + public boolean isConsumerTopic() { + return consumerGroup != null; + } + + public Collection getPartitionInfos() { + return partitionInfos; + } + + } + } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index ffec39078..6dc5c43b5 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -16,12 +16,13 @@ package org.springframework.cloud.stream.binder.kafka.config; -import javax.annotation.PostConstruct; import java.io.IOException; import java.util.HashMap; import java.util.List; import java.util.Map; +import javax.annotation.PostConstruct; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.ConsumerConfig; @@ -30,6 +31,7 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.utils.AppInfoParser; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.actuate.endpoint.PublicMetrics; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration; @@ -38,6 +40,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator; import org.springframework.cloud.stream.binder.kafka.KafkaBinderJaasInitializerListener; +import org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation; @@ -69,11 +72,13 @@ import org.springframework.util.ObjectUtils; * @author Soby Chacko * @author Mark Fisher * @author Ilayaperumal Gopinathan + * @author Henryk Konsek */ @Configuration @ConditionalOnMissingBean(Binder.class) -@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class, KafkaBinderConfiguration.KafkaPropertiesConfiguration.class}) -@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class}) +@Import({ KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class, + KafkaBinderConfiguration.KafkaPropertiesConfiguration.class }) +@EnableConfigurationProperties({ KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class }) public class KafkaBinderConfiguration { protected static final Log logger = LogFactory.getLog(KafkaBinderConfiguration.class); @@ -93,7 +98,7 @@ public class KafkaBinderConfiguration { @Autowired private ApplicationContext context; - @Autowired (required = false) + @Autowired(required = false) private AdminUtilsOperation adminUtilsOperation; @Bean @@ -132,6 +137,11 @@ public class KafkaBinderConfiguration { return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, consumerFactory); } + @Bean + public PublicMetrics kafkaBinderMetrics(KafkaMessageChannelBinder kafkaMessageChannelBinder) { + return new KafkaBinderMetrics(kafkaMessageChannelBinder, configurationProperties); + } + @Bean(name = "adminUtilsOperation") @Conditional(Kafka09Present.class) @ConditionalOnClass(name = "kafka.admin.AdminUtils") @@ -160,7 +170,7 @@ public class KafkaBinderConfiguration { return AppInfoParser.getVersion().startsWith("0.10"); } } - + static class Kafka09Present implements Condition { @Override @@ -195,24 +205,29 @@ public class KafkaBinderConfiguration { configuration.put(properties.getKey(), properties.getValue()); } } - for (Map.Entry producerProperties : this.kafkaProperties.buildProducerProperties().entrySet()) { + for (Map.Entry producerProperties : this.kafkaProperties.buildProducerProperties() + .entrySet()) { if (!configuration.containsKey(producerProperties.getKey())) { configuration.put(producerProperties.getKey(), producerProperties.getValue()); } } - for (Map.Entry consumerProperties : this.kafkaProperties.buildConsumerProperties().entrySet()) { + for (Map.Entry consumerProperties : this.kafkaProperties.buildConsumerProperties() + .entrySet()) { if (!configuration.containsKey(consumerProperties.getKey())) { configuration.put(consumerProperties.getKey(), consumerProperties.getValue()); } } if (ObjectUtils.isEmpty(configuration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { - configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBinderConfigurationProperties.getKafkaConnectionString()); + configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, + kafkaBinderConfigurationProperties.getKafkaConnectionString()); } else { @SuppressWarnings("unchecked") - List bootStrapServers = (List) configuration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG); + List bootStrapServers = (List) configuration + .get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG); if (bootStrapServers.size() == 1 && bootStrapServers.get(0).equals("localhost:9092")) { - configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBinderConfigurationProperties.getKafkaConnectionString()); + configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, + kafkaBinderConfigurationProperties.getKafkaConnectionString()); } } } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java index fcb9c6eb3..aba1cc370 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java @@ -49,7 +49,8 @@ import static org.junit.Assert.assertTrue; * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = {KafkaBinderAutoConfigurationPropertiesTest.KafkaBinderConfigProperties.class, KafkaBinderConfiguration.class}) +@SpringBootTest(classes = { KafkaBinderAutoConfigurationPropertiesTest.KafkaBinderConfigProperties.class, + KafkaBinderConfiguration.class }) @TestPropertySource(locations = "classpath:binder-config-autoconfig.properties") public class KafkaBinderAutoConfigurationPropertiesTest { @@ -62,13 +63,18 @@ public class KafkaBinderAutoConfigurationPropertiesTest { @Test public void testKafkaBinderConfigurationWithKafkaProperties() throws Exception { assertNotNull(this.kafkaMessageChannelBinder); - ExtendedProducerProperties producerProperties = new ExtendedProducerProperties<>(new KafkaProducerProperties()); - Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", ExtendedProducerProperties.class); + ExtendedProducerProperties producerProperties = new ExtendedProducerProperties<>( + new KafkaProducerProperties()); + Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", + ExtendedProducerProperties.class); getProducerFactoryMethod.setAccessible(true); - DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod.invoke(this.kafkaMessageChannelBinder, producerProperties); - Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs", Map.class); + DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod + .invoke(this.kafkaMessageChannelBinder, producerProperties); + Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs", + Map.class); ReflectionUtils.makeAccessible(producerFactoryConfigField); - Map producerConfigs = (Map) ReflectionUtils.getField(producerFactoryConfigField, producerFactory); + Map producerConfigs = (Map) ReflectionUtils.getField(producerFactoryConfigField, + producerFactory); assertTrue(producerConfigs.get("batch.size").equals(10)); assertTrue(producerConfigs.get("key.serializer").equals(LongSerializer.class)); assertTrue(producerConfigs.get("value.serializer").equals(LongSerializer.class)); @@ -77,13 +83,18 @@ public class KafkaBinderAutoConfigurationPropertiesTest { bootstrapServers.add("10.98.09.199:9092"); bootstrapServers.add("10.98.09.196:9092"); assertTrue((((List) producerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers))); - Method createKafkaConsumerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("createKafkaConsumerFactory", boolean.class, String.class, ExtendedConsumerProperties.class); + Method createKafkaConsumerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod( + "createKafkaConsumerFactory", boolean.class, String.class, ExtendedConsumerProperties.class); createKafkaConsumerFactoryMethod.setAccessible(true); - ExtendedConsumerProperties consumerProperties = new ExtendedConsumerProperties<>(new KafkaConsumerProperties()); - DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) createKafkaConsumerFactoryMethod.invoke(this.kafkaMessageChannelBinder, true, "test", consumerProperties); - Field consumerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", Map.class); + ExtendedConsumerProperties consumerProperties = new ExtendedConsumerProperties<>( + new KafkaConsumerProperties()); + DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) createKafkaConsumerFactoryMethod + .invoke(this.kafkaMessageChannelBinder, true, "test", consumerProperties); + Field consumerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", + Map.class); ReflectionUtils.makeAccessible(consumerFactoryConfigField); - Map consumerConfigs = (Map) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory); + Map consumerConfigs = (Map) ReflectionUtils.getField(consumerFactoryConfigField, + consumerFactory); assertTrue(consumerConfigs.get("key.deserializer").equals(LongDeserializer.class)); assertTrue(consumerConfigs.get("value.deserializer").equals(LongDeserializer.class)); assertTrue(consumerConfigs.get("group.id").equals("test")); @@ -106,7 +117,7 @@ public class KafkaBinderAutoConfigurationPropertiesTest { List bootstrapServers = new ArrayList<>(); bootstrapServers.add("10.98.09.199:9092"); bootstrapServers.add("10.98.09.196:9092"); - assertTrue(((List)configs.get("bootstrap.servers")).containsAll(bootstrapServers)); + assertTrue(((List) configs.get("bootstrap.servers")).containsAll(bootstrapServers)); } public static class KafkaBinderConfigProperties { @@ -115,5 +126,6 @@ public class KafkaBinderAutoConfigurationPropertiesTest { KafkaProperties kafkaProperties() { return new KafkaProperties(); } + } } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java index f3521f7d7..aa02c3490 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java @@ -46,7 +46,7 @@ import static org.junit.Assert.assertTrue; * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = {KafkaBinderConfiguration.class}) +@SpringBootTest(classes = { KafkaBinderConfiguration.class, KafkaBinderConfigurationPropertiesTest.class }) @TestPropertySource(locations = "classpath:binder-config.properties") public class KafkaBinderConfigurationPropertiesTest { @@ -60,13 +60,18 @@ public class KafkaBinderConfigurationPropertiesTest { kafkaProducerProperties.setBufferSize(12345); kafkaProducerProperties.setBatchTimeout(100); kafkaProducerProperties.setCompressionType(KafkaProducerProperties.CompressionType.gzip); - ExtendedProducerProperties producerProperties = new ExtendedProducerProperties<>(kafkaProducerProperties); - Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", ExtendedProducerProperties.class); + ExtendedProducerProperties producerProperties = new ExtendedProducerProperties<>( + kafkaProducerProperties); + Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", + ExtendedProducerProperties.class); getProducerFactoryMethod.setAccessible(true); - DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod.invoke(this.kafkaMessageChannelBinder, producerProperties); - Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs", Map.class); + DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod + .invoke(this.kafkaMessageChannelBinder, producerProperties); + Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs", + Map.class); ReflectionUtils.makeAccessible(producerFactoryConfigField); - Map producerConfigs = (Map) ReflectionUtils.getField(producerFactoryConfigField, producerFactory); + Map producerConfigs = (Map) ReflectionUtils.getField(producerFactoryConfigField, + producerFactory); assertTrue(producerConfigs.get("batch.size").equals("12345")); assertTrue(producerConfigs.get("linger.ms").equals("100")); assertTrue(producerConfigs.get("key.serializer").equals(ByteArraySerializer.class)); @@ -75,15 +80,21 @@ public class KafkaBinderConfigurationPropertiesTest { List bootstrapServers = new ArrayList<>(); bootstrapServers.add("10.98.09.199:9082"); assertTrue((((String) producerConfigs.get("bootstrap.servers")).contains("10.98.09.199:9082"))); - Method createKafkaConsumerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("createKafkaConsumerFactory", boolean.class, String.class, ExtendedConsumerProperties.class); + Method createKafkaConsumerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod( + "createKafkaConsumerFactory", boolean.class, String.class, ExtendedConsumerProperties.class); createKafkaConsumerFactoryMethod.setAccessible(true); - ExtendedConsumerProperties consumerProperties = new ExtendedConsumerProperties<>(new KafkaConsumerProperties()); - DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) createKafkaConsumerFactoryMethod.invoke(this.kafkaMessageChannelBinder, true, "test", consumerProperties); - Field consumerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", Map.class); + ExtendedConsumerProperties consumerProperties = new ExtendedConsumerProperties<>( + new KafkaConsumerProperties()); + DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) createKafkaConsumerFactoryMethod + .invoke(this.kafkaMessageChannelBinder, true, "test", consumerProperties); + Field consumerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", + Map.class); ReflectionUtils.makeAccessible(consumerFactoryConfigField); - Map consumerConfigs = (Map) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory); + Map consumerConfigs = (Map) ReflectionUtils.getField(consumerFactoryConfigField, + consumerFactory); assertTrue(consumerConfigs.get("key.deserializer").equals(ByteArrayDeserializer.class)); assertTrue(consumerConfigs.get("value.deserializer").equals(ByteArrayDeserializer.class)); assertTrue((((String) consumerConfigs.get("bootstrap.servers")).contains("10.98.09.199:9082"))); } + } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java index 1f05de3a1..6ff911e9b 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java @@ -33,7 +33,7 @@ import static org.junit.Assert.assertNotNull; * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = KafkaBinderConfiguration.class) +@SpringBootTest(classes = { KafkaBinderConfiguration.class, KafkaBinderConfigurationTest.class }) public class KafkaBinderConfigurationTest { @Autowired @@ -50,4 +50,5 @@ public class KafkaBinderConfigurationTest { producerListenerField, this.kafkaMessageChannelBinder); assertNotNull(producerListener); } + } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java index c00591cae..6e7250341 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java @@ -15,14 +15,11 @@ */ package org.springframework.cloud.stream.binder.kafka; -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.BDDMockito.given; - import java.util.ArrayList; -import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; + import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.Node; import org.apache.kafka.common.PartitionInfo; @@ -30,16 +27,21 @@ import org.junit.Before; import org.junit.Test; import org.mockito.Mock; import org.mockito.MockitoAnnotations; + import org.springframework.boot.actuate.health.Health; import org.springframework.boot.actuate.health.Status; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; + /** * @author Barry Commins */ public class KafkaBinderHealthIndicatorTest { private static final String TEST_TOPIC = "test"; + private KafkaBinderHealthIndicator indicator; @Mock @@ -51,7 +53,7 @@ public class KafkaBinderHealthIndicatorTest { @Mock private KafkaMessageChannelBinder binder; - private Map> topicsInUse = new HashMap<>(); + private Map topicsInUse = new HashMap<>(); @Before public void setup() { @@ -64,7 +66,7 @@ public class KafkaBinderHealthIndicatorTest { @Test public void kafkaBinderIsUp() { final List partitions = partitions(new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, partitions); + topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group", partitions)); given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.UP); @@ -73,7 +75,7 @@ public class KafkaBinderHealthIndicatorTest { @Test public void kafkaBinderIsDown() { final List partitions = partitions(new Node(-1, null, 0)); - topicsInUse.put(TEST_TOPIC, partitions); + topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group", partitions)); given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.DOWN); 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 new file mode 100644 index 000000000..1814d01bb --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java @@ -0,0 +1,137 @@ +/* + * Copyright 2017 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; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.TopicPartition; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import org.springframework.boot.actuate.metrics.Metric; +import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.TopicInformation; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; + +import static java.util.Collections.singletonMap; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyCollectionOf; +import static org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.METRIC_PREFIX; + +/** + * @author Henryk Konsek + */ +public class KafkaBinderMetricsTest { + + private static final String TEST_TOPIC = "test"; + + private KafkaBinderMetrics metrics; + + @Mock + private DefaultKafkaConsumerFactory consumerFactory; + + @Mock + private KafkaConsumer consumer; + + @Mock + private KafkaMessageChannelBinder binder; + + private Map topicsInUse = new HashMap<>(); + + @Mock + private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties; + + @Before + public void setup() { + MockitoAnnotations.initMocks(this); + given(consumerFactory.createConsumer()).willReturn(consumer); + given(binder.getTopicsInUse()).willReturn(topicsInUse); + metrics = new KafkaBinderMetrics(binder, kafkaBinderConfigurationProperties, consumerFactory); + given(consumer.endOffsets(anyCollectionOf(TopicPartition.class))) + .willReturn(singletonMap(new TopicPartition(TEST_TOPIC, 0), 1000L)); + } + + @Test + public void shouldIndicateLag() { + given(consumer.committed(any(TopicPartition.class))).willReturn(new OffsetAndMetadata(500)); + List partitions = partitions(new Node(0, null, 0)); + topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions)); + given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); + Collection> collectedMetrics = metrics.metrics(); + assertThat(collectedMetrics).hasSize(1); + assertThat(collectedMetrics.iterator().next().getName()) + .isEqualTo(String.format("%s.%s.%s.lag", METRIC_PREFIX, "group", TEST_TOPIC)); + assertThat(collectedMetrics.iterator().next().getValue()).isEqualTo(500L); + } + + @Test + public void shouldSumUpPartitionsLags() { + Map endOffsets = new HashMap<>(); + endOffsets.put(new TopicPartition(TEST_TOPIC, 0), 1000L); + endOffsets.put(new TopicPartition(TEST_TOPIC, 1), 1000L); + given(consumer.endOffsets(anyCollectionOf(TopicPartition.class))).willReturn(endOffsets); + given(consumer.committed(any(TopicPartition.class))).willReturn(new OffsetAndMetadata(500)); + List partitions = partitions(new Node(0, null, 0), new Node(0, null, 0)); + topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions)); + given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); + Collection> collectedMetrics = metrics.metrics(); + assertThat(collectedMetrics).hasSize(1); + assertThat(collectedMetrics.iterator().next().getName()) + .isEqualTo(String.format("%s.%s.%s.lag", METRIC_PREFIX, "group", TEST_TOPIC)); + assertThat(collectedMetrics.iterator().next().getValue()).isEqualTo(1000L); + } + + @Test + public void shouldIndicateFullLagForNotCommittedGroups() { + List partitions = partitions(new Node(0, null, 0)); + topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions)); + given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); + Collection> collectedMetrics = metrics.metrics(); + assertThat(collectedMetrics).hasSize(1); + assertThat(collectedMetrics.iterator().next().getName()) + .isEqualTo(String.format("%s.%s.%s.lag", METRIC_PREFIX, "group", TEST_TOPIC)); + assertThat(collectedMetrics.iterator().next().getValue()).isEqualTo(1000L); + } + + @Test + public void shouldNotCalculateLagForProducerTopics() { + List partitions = partitions(new Node(0, null, 0)); + topicsInUse.put(TEST_TOPIC, new TopicInformation(null, partitions)); + Collection> collectedMetrics = metrics.metrics(); + assertThat(collectedMetrics).isEmpty(); + } + + private List partitions(Node... nodes) { + List partitions = new ArrayList<>(); + for (int i = 0; i < nodes.length; i++) { + partitions.add(new PartitionInfo(TEST_TOPIC, i, nodes[i], null, null)); + } + return partitions; + } + +}