Kafka Streams binder metrics

Export Kafka Streams metrics available through KafkaStreams#metrics into a Micrometer MeterRegistry.
Add documentation for how to access metrics.
Modify test to verify metrics.

Resolves #543

docs/src/main/asciidoc/kafka-streams.adoc

@@ -1063,4 +1063,25 @@ Here is an example of using custom state stores with functional style described
 		}
 ----
 
-These state stores can be then accessed by the applications directly.
+These state stores can be then accessed by the applications directly.
+
+==== Accessing Kafka Streams Metrics
+
+Spring Cloud Stream Kafka Streams binder provides a basic mechanism for accessing Kafka Streams metrics exported through a MircoMeter `MeterRegistry`.
+Kafka Streams metrics that are available through `KafkaStreams#metrics()` are exported to this meter registry by the binder.
+The metrics exported are from the consumers, producers, admin-client and the stream itself.
+
+The metrics exported by the binder are exported with the format of metrics group name followed by a dot and then the actual metric name.
+All dashes in the original metric information is replaced with dots.
+
+For e.g. the metric name `network-io-total` from the metric group `consumer-metrics` is available in the micrometer registry as `consumer.metrics.network.io.total`.
+Similarly, the metric `commit-total` from `stream-metrics` is available as `stream.metrics.commit.total`.
+
+You can either programmatically access the Micrometer `MeterRegistry` in the application and then iterate through the available gauges or use Spring Boot actuator to access the metrics through a REST endpoint.
+When accessing through the Boot actuator endpoint, make sure to add `metrics` to the property `management.endpoints.web.exposure.include`.
+Then you can access `/acutator/metrics` to get a list of all the available metrics which then can be individually accessed through the same URL (`/actuator/metrics/<metric-name>`).
+
+Anything beyond the info level metrics available through `KafkaStreams#metrics()`, (for e.g. the debugging level metrics) are still only available through JMX after you set the `metrics.recording.level` to `DEBUG`.
+Kafka Streams, by default, set this level to `INFO`.
+https://kafka.apache.org/documentation/#kafka_streams_monitoring[Please see this section] from Kafka Streams documentation for more details.
+In a future release, binder may support exporting these DEBUG level metrics as well through Micrometer.

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2019-2019 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
+ *
+ *      https://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.streams;
+
+import java.util.Map;
+import java.util.function.ToDoubleFunction;
+
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.binder.MeterBinder;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.streams.KafkaStreams;
+
+/**
+ * Kafka Streams binder metrics implementation that exports the metrics available
+ * through {@link KafkaStreams#metrics()} into a micrometer {@link io.micrometer.core.instrument.MeterRegistry}.
+ *
+ * @author Soby Chacko
+ * @since 3.0.0
+ */
+public class KafkaStreamsBinderMetrics {
+
+	private KafkaStreams kafkaStreams;
+
+	private final MeterRegistry meterRegistry;
+
+	private MeterBinder meterBinder;
+
+	public KafkaStreamsBinderMetrics(MeterRegistry meterRegistry) {
+		this.meterRegistry = meterRegistry;
+	}
+
+	public void bindTo(MeterRegistry meterRegistry) {
+		if (this.meterBinder == null) {
+			this.meterBinder = new MeterBinder() {
+				@Override
+				@SuppressWarnings("unchecked")
+				public void bindTo(MeterRegistry registry) {
+					if (KafkaStreamsBinderMetrics.this.kafkaStreams != null) {
+						final Map<MetricName, ? extends Metric> metrics = KafkaStreamsBinderMetrics.this.kafkaStreams.metrics();
+
+						for (Map.Entry<MetricName, ? extends Metric> metric : metrics.entrySet()) {
+							final Gauge.Builder<KafkaStreamsBinderMetrics> builder =
+									Gauge.builder(sanitize(metric.getKey().group() + "." + metric.getKey().name()), this,
+											toDoubleFunction(metric.getValue()));
+							final Map<String, String> tags = metric.getKey().tags();
+							for (Map.Entry<String, String> tag : tags.entrySet()) {
+								builder.tag(tag.getKey(), tag.getValue());
+							}
+							builder.description(metric.getKey().description())
+									.register(meterRegistry);
+						}
+					}
+				}
+
+				ToDoubleFunction toDoubleFunction(Metric metric) {
+					return (o) -> {
+						if (metric.metricValue() instanceof Number) {
+							return (Double) metric.metricValue();
+						}
+						else {
+							return 0.0;
+						}
+					};
+				}
+			};
+		}
+		this.meterBinder.bindTo(this.meterRegistry);
+	}
+
+	public void addMetrics(KafkaStreams kafkaStreams) {
+		synchronized (KafkaStreamsBinderMetrics.this) {
+			this.kafkaStreams = kafkaStreams;
+			this.bindTo(this.meterRegistry);
+		}
+	}
+
+	private static String sanitize(String value) {
+		return value.replaceAll("-", ".");
+	}
+}

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java

@@ -23,6 +23,7 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Collectors;
 
+import io.micrometer.core.instrument.MeterRegistry;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
@@ -32,6 +33,7 @@ import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -48,8 +50,11 @@ import org.springframework.cloud.stream.config.BinderProperties;
 import org.springframework.cloud.stream.config.BindingServiceConfiguration;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.function.StreamFunctionProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.core.env.Environment;
 import org.springframework.core.env.MapPropertySource;
@@ -67,6 +72,7 @@ import org.springframework.util.StringUtils;
  * @author Soby Chacko
  * @author Gary Russell
  */
+@Configuration
 @EnableConfigurationProperties(KafkaStreamsExtendedBindingProperties.class)
 @ConditionalOnBean(BindingService.class)
 @AutoConfigureAfter(BindingServiceConfiguration.class)
@@ -338,8 +344,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
 	}
 
 	@Bean
-	public KafkaStreamsRegistry kafkaStreamsRegistry() {
-		return new KafkaStreamsRegistry();
+	public KafkaStreamsRegistry kafkaStreamsRegistry(KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics) {
+		return new KafkaStreamsRegistry(kafkaStreamsBinderMetrics);
 	}
 
 	@Bean
@@ -364,4 +370,35 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
 				cleanupConfig.getIfUnique(), streamFunctionProperties, kafkaStreamsBinderConfigurationProperties);
 	}
 
+
+
+	@Configuration
+	@ConditionalOnMissingBean(value = KafkaStreamsBinderMetrics.class, name = "outerContext")
+	@ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry")
+	protected class KafkaStreamsBinderMetricsConfiguration {
+
+		@Bean
+		@ConditionalOnBean(MeterRegistry.class)
+		@ConditionalOnMissingBean(KafkaStreamsBinderMetrics.class)
+		public KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics(
+				MeterRegistry meterRegistry) {
+
+			return new KafkaStreamsBinderMetrics(meterRegistry);
+		}
+	}
+
+	@Configuration
+	@ConditionalOnBean(name = "outerContext")
+	@ConditionalOnMissingBean(KafkaStreamsBinderMetrics.class)
+	@ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry")
+	protected class KafkaStreamsBinderMetricsConfigurationWithMultiBinder {
+
+		@Bean
+		public KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics(ConfigurableApplicationContext context) {
+
+			MeterRegistry meterRegistry = context.getBean("outerContext", ApplicationContext.class)
+					.getBean(MeterRegistry.class);
+			return new KafkaStreamsBinderMetrics(meterRegistry);
+		}
+	}
 }

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java

@@ -29,6 +29,12 @@ import org.apache.kafka.streams.KafkaStreams;
  */
 class KafkaStreamsRegistry {
 
+	private final KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics;
+
+	KafkaStreamsRegistry(KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics) {
+		this.kafkaStreamsBinderMetrics = kafkaStreamsBinderMetrics;
+	}
+
 	private final Set<KafkaStreams> kafkaStreams = new HashSet<>();
 
 	Set<KafkaStreams> getKafkaStreams() {
@@ -40,6 +46,7 @@ class KafkaStreamsRegistry {
 	 * @param kafkaStreams {@link KafkaStreams} object created in the application
 	 */
 	void registerKafkaStreams(KafkaStreams kafkaStreams) {
+		this.kafkaStreamsBinderMetrics.addMetrics(kafkaStreams);
 		this.kafkaStreams.add(kafkaStreams);
 	}
 

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java

@@ -95,7 +95,8 @@ public class KafkaStreamsInteractiveQueryIntegrationTests {
 	public void testStateStoreRetrievalRetry() {
 
 		KafkaStreams mock = Mockito.mock(KafkaStreams.class);
-		KafkaStreamsRegistry kafkaStreamsRegistry = new KafkaStreamsRegistry();
+		KafkaStreamsBinderMetrics mockMetrics = Mockito.mock(KafkaStreamsBinderMetrics.class);
+		KafkaStreamsRegistry kafkaStreamsRegistry = new KafkaStreamsRegistry(mockMetrics);
 		kafkaStreamsRegistry.registerKafkaStreams(mock);
 		KafkaStreamsBinderConfigurationProperties binderConfigurationProperties =
 				new KafkaStreamsBinderConfigurationProperties(new KafkaProperties());
@@ -140,8 +141,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests {
 		}
 	}
 
-	private void receiveAndValidateFoo(ConfigurableApplicationContext context)
-			throws Exception {
+	private void receiveAndValidateFoo(ConfigurableApplicationContext context) {
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
 				senderProps);

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java

@@ -21,6 +21,7 @@ import java.util.Date;
 import java.util.Map;
 import java.util.function.Function;
 
+import io.micrometer.core.instrument.MeterRegistry;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -35,9 +36,11 @@ import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -92,6 +95,8 @@ public class KafkaStreamsBinderWordCountFunctionTests {
 						"=org.apache.kafka.common.serialization.Serdes$StringSerde",
 				"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
 			receiveAndValidate("words", "counts");
+			final MeterRegistry meterRegistry = context.getBean(MeterRegistry.class);
+			assertThat(meterRegistry.get("stream.metrics.commit.total").gauge().value()).isEqualTo(1.0);
 		}
 	}
 
@@ -183,6 +188,9 @@ public class KafkaStreamsBinderWordCountFunctionTests {
 	@EnableAutoConfiguration
 	public static class WordCountProcessorApplication {
 
+		@Autowired
+		InteractiveQueryService interactiveQueryService;
+
 		@Bean
 		public Function<KStream<Object, String>, KStream<?, WordCount>> process() {