spring-projectsGH-2198: Observablility Documentation, Polish

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

spring-projectsGH-2417: Fix class tangles.

Resolves spring-projects#2417

Disable auto doc generation and polish manually.

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,8 +67,9 @@ 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'
 	scalaVersion = '2.13'
@@ -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/KafkaTemplate.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.function.Supplier;
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -64,9 +65,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 +146,7 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private KafkaTemplateObservationConvention observationConvention;
 
-	private ObservationRegistry observationRegistry;
+	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
 
 	/**
 	 * Create an instance using the supplied producer factory and autoFlush false.
@@ -418,7 +419,7 @@ public void setObservationConvention(KafkaTemplateObservationConvention observat
 
 	@Override
 	public void afterSingletonsInstantiated() {
-		if (this.observationEnabled && this.observationRegistry == null && this.applicationContext != null) {
+		if (this.observationEnabled && this.applicationContext != null) {
 			ObjectProvider<ObservationRegistry> registry =
 					this.applicationContext.getBeanProvider(ObservationRegistry.class);
 			this.observationRegistry = registry.getIfUnique();
@@ -668,15 +669,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,
+				(Supplier<KafkaRecordSenderContext>) () -> new KafkaRecordSenderContext(producerRecord, this.beanName),
+				this.observationRegistry);
 		try {
 			observation.start();
 			return doSend(producerRecord, observation);

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -43,6 +43,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
+import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -107,8 +108,8 @@
 import org.springframework.kafka.support.LogIfLevelEnabled;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.TopicPartitionOffset.SeekPosition;
-import org.springframework.kafka.support.micrometer.DefaultKafkaListenerObservationConvention;
 import org.springframework.kafka.support.micrometer.KafkaListenerObservation;
+import org.springframework.kafka.support.micrometer.KafkaListenerObservation.DefaultKafkaListenerObservationConvention;
 import org.springframework.kafka.support.micrometer.KafkaRecordReceiverContext;
 import org.springframework.kafka.support.micrometer.MicrometerHolder;
 import org.springframework.kafka.support.serializer.DeserializationException;
@@ -365,9 +366,9 @@ protected void doStart() {
 		}
 		GenericMessageListener<?> listener = (GenericMessageListener<?>) messageListener;
 		ListenerType listenerType = determineListenerType(listener);
-		ObservationRegistry observationRegistry = null;
+		ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
 		ApplicationContext applicationContext = getApplicationContext();
-		if (applicationContext != null) {
+		if (applicationContext != null && containerProperties.isObservationEnabled()) {
 			ObjectProvider<ObservationRegistry> registry =
 					applicationContext.getBeanProvider(ObservationRegistry.class);
 			observationRegistry = registry.getIfUnique();
@@ -827,7 +828,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		@SuppressWarnings(UNCHECKED)
 		ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType,
-				@Nullable ObservationRegistry observationRegistry) {
+				ObservationRegistry observationRegistry) {
 
 			this.observationRegistry = observationRegistry;
 			Properties consumerProperties = propertiesFromProperties();
@@ -2706,16 +2707,12 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 				Iterator<ConsumerRecord<K, V>> iterator) {
 
 			Object sample = startMicrometerSample();
-			Observation observation;
-			if (!this.containerProperties.isObservationEnabled() || this.observationRegistry == null) {
-				observation = Observation.NOOP;
-			}
-			else {
-				observation = KafkaListenerObservation.LISTENER_OBSERVATION.observation(
-						this.containerProperties.getObservationConvention(),
-						DefaultKafkaListenerObservationConvention.INSTANCE,
-						new KafkaRecordReceiverContext(record, getListenerId()), this.observationRegistry);
-			}
+			Observation observation = KafkaListenerObservation.LISTENER_OBSERVATION.observation(
+					this.containerProperties.getObservationConvention(),
+					DefaultKafkaListenerObservationConvention.INSTANCE,
+					(Supplier<KafkaRecordReceiverContext>) () -> new KafkaRecordReceiverContext(record,
+							getListenerId()),
+					this.observationRegistry);
 			return observation.observe(() -> {
 				try {
 					invokeOnMessage(record);