GH-2198: Observability Documentation

Resolves #2417

Refactor for latest snapshots; add documentation generation.
Use NOOP registry and supplier for context.

GH-2417: Fix class tangles.

Disable auto doc generation and polish manually.

Remove unnecessary `Supplier` casts.

Remove incorrect `@Nullable`.

Add Remote Service Name to Contexts

Include the cluster id if possible.

* Move deps to the latest snapshots

Author: garyrussell

build.gradle

@@ -1,5 +1,6 @@
 buildscript {
 	ext.kotlinVersion = '1.7.0'
+	ext.isCI = System.getenv('GITHUB_ACTION') || System.getenv('bamboo_buildKey')
 	repositories {
 		mavenCentral()
 		gradlePluginPortal()
@@ -66,15 +67,16 @@ ext {
 	junitJupiterVersion = '5.9.0'
 	kafkaVersion = '3.3.1'
 	log4jVersion = '2.18.0'
-	micrometerVersion = '1.10.0-M6'
-	micrometerTracingVersion = '1.0.0-M8'
+	micrometerDocsVersion = "1.0.0-SNAPSHOT"
+	micrometerVersion = '1.10.0-SNAPSHOT'
+	micrometerTracingVersion = '1.0.0-SNAPSHOT'
 	mockitoVersion = '4.5.1'
-	reactorVersion = '2022.0.0-M6'
+	reactorVersion = '2022.0.0-SNAPSHOT'
 	scalaVersion = '2.13'
 	springBootVersion = '2.6.11' // docs module
-	springDataVersion = '2022.0.0-M6'
-	springRetryVersion = '2.0.0-M1'
-	springVersion = '6.0.0-M6'
+	springDataVersion = '2022.0.0-SNAPSHOT'
+	springRetryVersion = '2.0.0-SNAPSHOT'
+	springVersion = '6.0.0-SNAPSHOT'
 	zookeeperVersion = '3.6.3'
 
 	idPrefix = 'kafka'
@@ -241,7 +243,7 @@ subprojects { subproject ->
 	}
 
 	task updateCopyrights {
-		onlyIf { gitPresent && !System.getenv('GITHUB_ACTION') && !System.getenv('bamboo_buildKey') }
+		onlyIf { !isCI }
 		if (gitPresent) {
 			inputs.files(modifiedFiles.filter { f -> f.path.contains(subproject.name) })
 		}
@@ -300,13 +302,17 @@ subprojects { subproject ->
 	tasks.withType(Javadoc) {
 		options.addBooleanOption('Xdoclint:syntax', true) // only check syntax with doclint
 		options.addBooleanOption('Werror', true) // fail build on Javadoc warnings
-    }
+	}
 
 }
 
 project ('spring-kafka') {
 	description = 'Spring Kafka Support'
 
+	configurations {
+		adoc
+	}
+
 	dependencies {
 		api 'org.springframework:spring-context'
 		api 'org.springframework:spring-messaging'
@@ -346,7 +352,36 @@ project ('spring-kafka') {
 		testImplementation 'io.micrometer:micrometer-tracing-bridge-brave'
 		testImplementation 'io.micrometer:micrometer-tracing-test'
 		testImplementation 'io.micrometer:micrometer-tracing-integration-test'
+
+		adoc "io.micrometer:micrometer-docs-generator-spans:$micrometerDocsVersion"
+		adoc "io.micrometer:micrometer-docs-generator-metrics:$micrometerDocsVersion"
 	}
+
+	def inputDir = file('src/main/java/org/springframework/kafka/support/micrometer').absolutePath
+	def outputDir = rootProject.file('spring-kafka-docs/src/main/asciidoc').absolutePath
+
+	task generateObservabilityMetricsDocs(type: JavaExec) {
+		onlyIf { !isCI }
+		mainClass = 'io.micrometer.docs.metrics.DocsFromSources'
+		inputs.dir(inputDir)
+		outputs.dir(outputDir)
+		classpath configurations.adoc
+		args inputDir, '.*', outputDir
+	}
+
+	task generateObservabilitySpansDocs(type: JavaExec) {
+		onlyIf { !isCI }
+		mainClass = 'io.micrometer.docs.spans.DocsFromSources'
+		inputs.dir(inputDir)
+		outputs.dir(outputDir)
+		classpath configurations.adoc
+		args inputDir, '.*', outputDir
+	}
+
+	// javadoc {
+	// 	finalizedBy generateObservabilityMetricsDocs, generateObservabilitySpansDocs
+	// }
+
 }
 
 project ('spring-kafka-test') {
@@ -564,7 +599,7 @@ task distZip(type: Zip, dependsOn: [docsZip]) { //, schemaZip]) {
 		into "${baseDir}"
 	}
 
-    from("$project.rootDir") {
+	from("$project.rootDir") {
 		include 'LICENSE.txt'
 		into "${baseDir}"
 	}

spring-kafka-docs/src/main/asciidoc/_conventions.adoc

@@ -0,0 +1,11 @@
+[[observability-conventions]]
+=== Observability - Conventions
+
+Below you can find a list of all `GlobalObservabilityConventions` and `ObservabilityConventions` declared by this project.
+
+.ObservationConvention implementations
+|===
+|ObservationConvention Class Name | Applicable ObservationContext Class Name
+|`KafkaListenerObservation$DefaultKafkaListenerObservationConvention`|`KafkaRecordReceiverContext`
+|`KafkaTemplateObservation$DefaultKafkaTemplateObservationConvention`|`KafkaRecordSenderContext`
+|===

spring-kafka-docs/src/main/asciidoc/_metrics.adoc

@@ -0,0 +1,44 @@
+[[observability-metrics]]
+=== Observability - Metrics
+
+Below you can find a list of all samples declared by this project.
+
+[[observability-metrics-listener-observation]]
+==== Listener Observation
+
+____
+Observation for Apache Kafka listeners.
+____
+
+**Metric name** `spring.kafka.listener` (defined by convention class `KafkaListenerObservation$DefaultKafkaListenerObservationConvention`). **Type** `timer` and **base unit** `seconds`.
+
+Name of the enclosing class `KafkaListenerObservation`.
+
+IMPORTANT: All tags must be prefixed with `spring.kafka.listener` prefix!
+
+.Low cardinality Keys
+[cols="a,a"]
+|===
+|Name | Description
+|`spring.kafka.listener.id`|Listener id (or listener container bean name).
+|===
+
+[[observability-metrics-template-observation]]
+==== Template Observation
+
+____
+Observation for KafkaTemplates.
+____
+
+**Metric name** `spring.kafka.template` (defined by convention class `KafkaTemplateObservation$DefaultKafkaTemplateObservationConvention`). **Type** `timer` and **base unit** `seconds`.
+
+Name of the enclosing class `KafkaTemplateObservation`.
+
+IMPORTANT: All tags must be prefixed with `spring.kafka.template` prefix!
+
+.Low cardinality Keys
+[cols="a,a"]
+|===
+|Name | Description
+|`spring.kafka.template.name`|Bean name of the template.
+|===

spring-kafka-docs/src/main/asciidoc/_spans.adoc

@@ -0,0 +1,38 @@
+[[observability-spans]]
+=== Observability - Spans
+
+Below you can find a list of all spans declared by this project.
+
+[[observability-spans-listener-observation]]
+==== Listener Observation Span
+
+> Observation for Apache Kafka listeners.
+
+**Span name** `spring.kafka.listener` (defined by convention class `KafkaListenerObservation$DefaultKafkaListenerObservationConvention`).
+
+Name of the enclosing class `KafkaListenerObservation`.
+
+IMPORTANT: All tags and event names must be prefixed with `spring.kafka.listener` prefix!
+
+.Tag Keys
+|===
+|Name | Description
+|`spring.kafka.listener.id`|Listener id (or listener container bean name).
+|===
+
+[[observability-spans-template-observation]]
+==== Template Observation Span
+
+> Observation for KafkaTemplates.
+
+**Span name** `spring.kafka.template` (defined by convention class `KafkaTemplateObservation$DefaultKafkaTemplateObservationConvention`).
+
+Name of the enclosing class `KafkaTemplateObservation`.
+
+IMPORTANT: All tags and event names must be prefixed with `spring.kafka.template` prefix!
+
+.Tag Keys
+|===
+|Name | Description
+|`spring.kafka.template.name`|Bean name of the template.
+|===

spring-kafka-docs/src/main/asciidoc/appendix.adoc

@@ -49,6 +49,16 @@ dependencies {
 
 The test scope dependencies are only needed if you are using the embedded Kafka broker in tests.
 
+[appendix]
+[[observation-gen]]
+== Micrometer Observation Documentation
+
+include::./_metrics.adoc[]
+
+include::./_spans.adoc[]
+
+include::./_conventions.adoc[]
+
 [appendix]
 [[history]]
 == Change History

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

@@ -3415,6 +3415,8 @@ The default implementations add the `bean.name` tag for template observations an
 
 You can either subclass `DefaultKafkaTemplateObservationConvention` or `DefaultKafkaListenerObservationConvention` or provide completely new implementations.
 
+See <<observation-gen>> for details of the observations that are recorded.
+
 [[transactions]]
 ==== Transactions
 

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java

@@ -57,6 +57,7 @@
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
+import org.springframework.lang.Nullable;
 
 /**
  * An admin that delegates to an {@link AdminClient} to create topics defined
@@ -95,6 +96,8 @@ public class KafkaAdmin extends KafkaResourceFactory
 
 	private boolean modifyTopicConfigs;
 
+	private String clusterId;
+
 	/**
 	 * Create an instance with an {@link AdminClient} based on the supplied
 	 * configuration.
@@ -197,6 +200,10 @@ public final boolean initialize() {
 			}
 			if (adminClient != null) {
 				try {
+					synchronized (this) {
+						this.clusterId = adminClient.describeCluster().clusterId().get(this.operationTimeout,
+								TimeUnit.MILLISECONDS);
+					}
 					addOrModifyTopicsIfNeeded(adminClient, newTopics);
 					return true;
 				}
@@ -218,6 +225,20 @@ public final boolean initialize() {
 		return false;
 	}
 
+	@Override
+	@Nullable
+	public String clusterId() {
+		if (this.clusterId == null) {
+			try (AdminClient client = createAdmin()) {
+				this.clusterId = client.describeCluster().clusterId().get(this.operationTimeout, TimeUnit.MILLISECONDS);
+			}
+			catch (Exception ex) {
+				LOGGER.error(ex, "Could not obtaine cluster info");
+			}
+		}
+		return this.clusterId;
+	}
+
 	@Override
 	public void createOrModifyTopics(NewTopic... topics) {
 		try (AdminClient client = createAdmin()) {

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdminOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -21,6 +21,8 @@
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 
+import org.springframework.lang.Nullable;
+
 /**
  * Provides a number of convenience methods wrapping {@code AdminClient}.
  *
@@ -49,4 +51,12 @@ public interface KafkaAdminOperations {
 	 */
 	Map<String, TopicDescription> describeTopics(String... topicNames);
 
+	/**
+	 * Return the cluster id, if available.
+	 * @return the describe cluster id.
+	 * @since 3.0
+	 */
+	@Nullable
+	String clusterId();
+
 }

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -48,7 +48,6 @@
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
-import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -64,9 +63,9 @@
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
-import org.springframework.kafka.support.micrometer.DefaultKafkaTemplateObservationConvention;
 import org.springframework.kafka.support.micrometer.KafkaRecordSenderContext;
 import org.springframework.kafka.support.micrometer.KafkaTemplateObservation;
+import org.springframework.kafka.support.micrometer.KafkaTemplateObservation.DefaultKafkaTemplateObservationConvention;
 import org.springframework.kafka.support.micrometer.KafkaTemplateObservationConvention;
 import org.springframework.kafka.support.micrometer.MicrometerHolder;
 import org.springframework.lang.Nullable;
@@ -145,7 +144,11 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private KafkaTemplateObservationConvention observationConvention;
 
-	private ObservationRegistry observationRegistry;
+	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
+
+	private KafkaAdmin kafkaAdmin;
+
+	private String clusterId;
 
 	/**
 	 * Create an instance using the supplied producer factory and autoFlush false.
@@ -418,16 +421,25 @@ public void setObservationConvention(KafkaTemplateObservationConvention observat
 
 	@Override
 	public void afterSingletonsInstantiated() {
-		if (this.observationEnabled && this.observationRegistry == null && this.applicationContext != null) {
-			ObjectProvider<ObservationRegistry> registry =
-					this.applicationContext.getBeanProvider(ObservationRegistry.class);
-			this.observationRegistry = registry.getIfUnique();
+		if (this.observationEnabled && this.applicationContext != null) {
+			this.observationRegistry = this.applicationContext.getBeanProvider(ObservationRegistry.class).getIfUnique();
+			this.kafkaAdmin = this.applicationContext.getBeanProvider(KafkaAdmin.class).getIfUnique();
+			if (this.kafkaAdmin != null) {
+				this.clusterId = this.kafkaAdmin.clusterId();
+			}
 		}
 		else if (this.micrometerEnabled) {
 			this.micrometerHolder = obtainMicrometerHolder();
 		}
 	}
 
+	private String clusterId() {
+		if (this.kafkaAdmin != null && this.clusterId == null) {
+			this.clusterId = this.kafkaAdmin.clusterId();
+		}
+		return this.clusterId;
+	}
+
 	@Override
 	public void onApplicationEvent(ContextStoppedEvent event) {
 		if (this.customProducerFactory) {
@@ -668,15 +680,10 @@ protected void closeProducer(Producer<K, V> producer, boolean inTx) {
 	}
 
 	private CompletableFuture<SendResult<K, V>> observeSend(final ProducerRecord<K, V> producerRecord) {
-		Observation observation;
-		if (!this.observationEnabled || this.observationRegistry == null) {
-			observation = Observation.NOOP;
-		}
-		else {
-			observation = KafkaTemplateObservation.TEMPLATE_OBSERVATION.observation(
-					this.observationConvention, DefaultKafkaTemplateObservationConvention.INSTANCE,
-					new KafkaRecordSenderContext(producerRecord, this.beanName), this.observationRegistry);
-		}
+		Observation observation = KafkaTemplateObservation.TEMPLATE_OBSERVATION.observation(
+				this.observationConvention, DefaultKafkaTemplateObservationConvention.INSTANCE,
+				() -> new KafkaRecordSenderContext(producerRecord, this.beanName, this::clusterId),
+				this.observationRegistry);
 		try {
 			observation.start();
 			return doSend(producerRecord, observation);
@@ -695,7 +702,7 @@ private CompletableFuture<SendResult<K, V>> observeSend(final ProducerRecord<K,
 	 * RecordMetadata}.
 	 */
 	protected CompletableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord,
-			@Nullable Observation observation) {
+			Observation observation) {
 
 		final Producer<K, V> producer = getTheProducer(producerRecord.topic());
 		this.logger.trace(() -> "Sending: " + KafkaUtils.format(producerRecord));