Reference doc update for Kafka

docs/modules/ROOT/nav.adoc

@@ -42,6 +42,7 @@
 ** xref:reference/httpcomponents.adoc[HttpComponents Client]
 ** xref:reference/jetty.adoc[Jetty and Jersey]
 ** xref:reference/jvm.adoc[JVM]
+** xref:reference/kafka.adoc[Kafka]
 ** xref:reference/netty.adoc[Netty]
 ** xref:reference/okhttpclient.adoc[OkHttpClient]
 * xref:guides.adoc[Guides]

docs/modules/ROOT/pages/reference/kafka.adoc

@@ -0,0 +1,31 @@
+[[overview]]
+= Apache Kafka Metrics
+
+https://kafka.apache.org/[Apache Kafka] is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.
+
+Below you can find an example of how to instrument Apache Kafka with Micrometer.
+
+Setting up instrumentation with Apache Kafka on the consumer side.
+
+[source,java,subs=+attributes]
+-----
+// Setting up and binding instrumentation
+include::{include-core-test-java}/io/micrometer/core/instrument/binder/kafka/KafkaClientMetricsIntegrationTest.java[tags=consumer_setup, indent=0]
+-----
+
+Setting up instrumentation with Apache Kafka on the producer side.
+
+[source,java,subs=+attributes]
+-----
+// Setting up and binding instrumentation
+include::{include-core-test-java}/io/micrometer/core/instrument/binder/kafka/KafkaClientMetricsIntegrationTest.java[tags=producer_setup, indent=0]
+-----
+
+Setting up instrumentation with Apache Kafka using Kafka Streams.
+
+[source,java,subs=+attributes]
+-----
+// Setting up, binding instrumentation and usage example
+include::{include-core-test-java}/io/micrometer/core/instrument/binder/kafka/KafkaStreamsMetricsTest.java[tags=example, indent=0]
+-----
+

micrometer-core/src/test/java/io/micrometer/core/instrument/binder/kafka/KafkaClientMetricsIntegrationTest.java

@@ -57,18 +57,21 @@ void shouldManageProducerAndConsumerMetrics() {
 
         assertThat(registry.getMeters()).hasSize(0);
 
+        // tag::producer_setup[]
         Properties producerConfigs = new Properties();
         producerConfigs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
         Producer<String, String> producer = new KafkaProducer<>(producerConfigs, new StringSerializer(),
                 new StringSerializer());
 
         KafkaClientMetrics producerKafkaMetrics = new KafkaClientMetrics(producer);
         producerKafkaMetrics.bindTo(registry);
+        // end::producer_setup[]
 
         int producerMetrics = registry.getMeters().size();
         assertThat(registry.getMeters()).hasSizeGreaterThan(0);
         assertThat(registry.getMeters()).extracting(m -> m.getId().getTag("kafka.version")).allMatch(v -> !v.isEmpty());
 
+        // tag::consumer_setup[]
         Properties consumerConfigs = new Properties();
         consumerConfigs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
         consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
@@ -77,6 +80,7 @@ void shouldManageProducerAndConsumerMetrics() {
 
         KafkaClientMetrics consumerKafkaMetrics = new KafkaClientMetrics(consumer);
         consumerKafkaMetrics.bindTo(registry);
+        // end::consumer_setup[]
 
         // Printing out for discovery purposes
         out.println("Meters from producer before sending:");

micrometer-core/src/test/java/io/micrometer/core/instrument/binder/kafka/KafkaStreamsMetricsTest.java

@@ -46,6 +46,7 @@ void afterEach() {
 
     @Test
     void shouldCreateMeters() {
+        // tag::example[]
         try (KafkaStreams kafkaStreams = createStreams()) {
             metrics = new KafkaStreamsMetrics(kafkaStreams);
             MeterRegistry registry = new SimpleMeterRegistry();
@@ -55,6 +56,7 @@ void shouldCreateMeters() {
                 .extracting(meter -> meter.getId().getName())
                 .allMatch(s -> s.startsWith(METRIC_NAME_PREFIX));
         }
+        // end::example[]
     }
 
     @Test