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.

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]

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.

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<PartitionInfo> 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());
 					}
 				}

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<Metric<?>> metrics() {
+		List<Metric<?>> metrics = new LinkedList<>();
+		for (Map.Entry<String, KafkaMessageChannelBinder.TopicInformation> 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<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);
+				List<TopicPartition> topicPartitions = new LinkedList<>();
+				for (PartitionInfo partitionInfo : partitionInfos) {
+					topicPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
+				}
+				Map<TopicPartition, Long> endOffsets = metadataConsumer.endOffsets(topicPartitions);
+				long lag = 0;
+				for (Map.Entry<TopicPartition, Long> 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<String, Object> 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);
+	}
+
+}

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<ExtendedConsumerProperties<KafkaConsumerProperties>,
-				ExtendedProducerProperties<KafkaProducerProperties>, KafkaTopicProvisioner>
+		AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>, KafkaTopicProvisioner>
 		implements ExtendedPropertiesBinder<MessageChannel, KafkaConsumerProperties, KafkaProducerProperties> {
 
 	private final KafkaBinderConfigurationProperties configurationProperties;
@@ -97,7 +96,7 @@ public class KafkaMessageChannelBinder extends
 
 	private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties();
 
-	private final Map<String, Collection<PartitionInfo>> topicsInUse = new HashMap<>();
+	private final Map<String, TopicInformation> topicsInUse = new HashMap<>();
 
 	public KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties,
 			KafkaTopicProvisioner provisioningProvider) {
@@ -129,7 +128,7 @@ public class KafkaMessageChannelBinder extends
 		this.producerListener = producerListener;
 	}
 
-	Map<String, Collection<PartitionInfo>> getTopicsInUse() {
+	Map<String, TopicInformation> getTopicsInUse() {
 		return this.topicsInUse;
 	}
 
@@ -147,7 +146,8 @@ public class KafkaMessageChannelBinder extends
 	protected MessageHandler createProducerMessageHandler(final ProducerDestination destination,
 			ExtendedProducerProperties<KafkaProducerProperties> producerProperties) throws Exception {
 		final DefaultKafkaProducerFactory<byte[], byte[]> producerFB = getProducerFactory(producerProperties);
-		Collection<PartitionInfo> partitions = provisioningProvider.getPartitionsForTopic(producerProperties.getPartitionCount(),
+		Collection<PartitionInfo> partitions = provisioningProvider.getPartitionsForTopic(
+				producerProperties.getPartitionCount(),
 				false,
 				new Callable<Collection<PartitionInfo>>() {
 					@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<byte[], byte[]> 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<PartitionInfo> 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<byte[], byte[]> producerFactory = getProducerFactory(new ExtendedProducerProperties<>(new KafkaProducerProperties()));
+			DefaultKafkaProducerFactory<byte[], byte[]> producerFactory = getProducerFactory(
+					new ExtendedProducerProperties<>(new KafkaProducerProperties()));
 			final KafkaTemplate<byte[], byte[]> 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<SendResult<byte[], byte[]>> sentDlq = kafkaTemplate.send(dlqName, message.partition(), key, payload);
+					String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName())
+							? extendedConsumerProperties.getExtension().getDlqName()
+							: "error." + destination.getName() + "." + group;
+					ListenableFuture<SendResult<byte[], byte[]>> sentDlq = kafkaTemplate.send(dlqName,
+							message.partition(), key, payload);
 					sentDlq.addCallback(new ListenableFutureCallback<SendResult<byte[], byte[]>>() {
 						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<PartitionInfo> 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<PartitionInfo> partitionInfos;
+
+		public TopicInformation(String consumerGroup, Collection<PartitionInfo> partitionInfos) {
+			this.consumerGroup = consumerGroup;
+			this.partitionInfos = partitionInfos;
+		}
+
+		public String getConsumerGroup() {
+			return consumerGroup;
+		}
+
+		public boolean isConsumerTopic() {
+			return consumerGroup != null;
+		}
+
+		public Collection<PartitionInfo> getPartitionInfos() {
+			return partitionInfos;
+		}
+
+	}
+
 }

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<String, Object> producerProperties : this.kafkaProperties.buildProducerProperties().entrySet()) {
+				for (Map.Entry<String, Object> producerProperties : this.kafkaProperties.buildProducerProperties()
+						.entrySet()) {
 					if (!configuration.containsKey(producerProperties.getKey())) {
 						configuration.put(producerProperties.getKey(), producerProperties.getValue());
 					}
 				}
-				for (Map.Entry<String, Object> consumerProperties : this.kafkaProperties.buildConsumerProperties().entrySet()) {
+				for (Map.Entry<String, Object> 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<String> bootStrapServers = (List<String>) configuration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
+					List<String> bootStrapServers = (List<String>) 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());
 					}
 				}
 			}

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<KafkaProducerProperties> producerProperties = new ExtendedProducerProperties<>(new KafkaProducerProperties());
-		Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", ExtendedProducerProperties.class);
+		ExtendedProducerProperties<KafkaProducerProperties> 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<String, Object> producerConfigs = (Map<String, Object>) ReflectionUtils.getField(producerFactoryConfigField, producerFactory);
+		Map<String, Object> producerConfigs = (Map<String, Object>) 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<String>) 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<KafkaConsumerProperties> 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<KafkaConsumerProperties> 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<String, Object> consumerConfigs = (Map<String, Object>) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory);
+		Map<String, Object> consumerConfigs = (Map<String, Object>) 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<String> bootstrapServers = new ArrayList<>();
 		bootstrapServers.add("10.98.09.199:9092");
 		bootstrapServers.add("10.98.09.196:9092");
-		assertTrue(((List<String>)configs.get("bootstrap.servers")).containsAll(bootstrapServers));
+		assertTrue(((List<String>) configs.get("bootstrap.servers")).containsAll(bootstrapServers));
 	}
 
 	public static class KafkaBinderConfigProperties {
@@ -115,5 +126,6 @@ public class KafkaBinderAutoConfigurationPropertiesTest {
 		KafkaProperties kafkaProperties() {
 			return new KafkaProperties();
 		}
+
 	}
 }

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<KafkaProducerProperties> producerProperties = new ExtendedProducerProperties<>(kafkaProducerProperties);
-		Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", ExtendedProducerProperties.class);
+		ExtendedProducerProperties<KafkaProducerProperties> 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<String, Object> producerConfigs = (Map<String, Object>) ReflectionUtils.getField(producerFactoryConfigField, producerFactory);
+		Map<String, Object> producerConfigs = (Map<String, Object>) 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<String> 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<KafkaConsumerProperties> 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<KafkaConsumerProperties> 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<String, Object> consumerConfigs = (Map<String, Object>) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory);
+		Map<String, Object> consumerConfigs = (Map<String, Object>) 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")));
 	}
+
 }

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);
 	}
+
 }

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<String, Collection<PartitionInfo>> topicsInUse = new HashMap<>();
+	private Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = new HashMap<>();
 
 	@Before
 	public void setup() {
@@ -64,7 +66,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Test
 	public void kafkaBinderIsUp() {
 		final List<PartitionInfo> 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<PartitionInfo> 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);

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<String, TopicInformation> 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<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions));
+		given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions);
+		Collection<Metric<?>> 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<TopicPartition, Long> 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<PartitionInfo> 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<Metric<?>> 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<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC, new TopicInformation("group", partitions));
+		given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions);
+		Collection<Metric<?>> 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<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(null, partitions));
+		Collection<Metric<?>> collectedMetrics = metrics.metrics();
+		assertThat(collectedMetrics).isEmpty();
+	}
+
+	private List<PartitionInfo> partitions(Node... nodes) {
+		List<PartitionInfo> partitions = new ArrayList<>();
+		for (int i = 0; i < nodes.length; i++) {
+			partitions.add(new PartitionInfo(TEST_TOPIC, i, nodes[i], null, null));
+		}
+		return partitions;
+	}
+
+}