GH-2638: Support Dynamic Tags via Observation

See #2648

Author: garyrussell

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

@@ -3447,7 +3447,11 @@ 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.
+See <<observation-gen>> for details of the default observations that are recorded.
+
+Starting with version 3.0.6, you can add dynamic tags to the timers and traces, based on information in the consumer or producer records.
+To do so, add a custom `KafkaListenerObservationConvention` and/or `KafkaTemplateObservationConvention` to the listener container properties or `KafkaTemplate` respectively.
+The `record` property in both observation contexts contains the `ConsumerRecord` or `ProducerRecord` respectively.
 
 [[transactions]]
 ==== Transactions

spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordReceiverContext.java

@@ -52,6 +52,10 @@ public KafkaRecordReceiverContext(ConsumerRecord<?, ?> record, String listenerId
 		setRemoteServiceName("Apache Kafka" + (cluster != null ? ": " + cluster : ""));
 	}
 
+	/**
+	 * Return the listener id.
+	 * @return the listener id.
+	 */
 	public String getListenerId() {
 		return this.listenerId;
 	}
@@ -64,4 +68,13 @@ public String getSource() {
 		return this.record.topic();
 	}
 
+	/**
+	 * Return the consumer record.
+	 * @return the record the record.
+	 * @since 3.0.6
+	 */
+	public ConsumerRecord<?, ?> getRecord() {
+		return this.record;
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordSenderContext.java

@@ -34,17 +34,21 @@ public class KafkaRecordSenderContext extends SenderContext<ProducerRecord<?, ?>
 
 	private final String beanName;
 
-	private final String destination;
+	private final ProducerRecord<?, ?> record;
 
 	public KafkaRecordSenderContext(ProducerRecord<?, ?> record, String beanName, Supplier<String> clusterId) {
 		super((carrier, key, value) -> record.headers().add(key, value.getBytes(StandardCharsets.UTF_8)));
 		setCarrier(record);
 		this.beanName = beanName;
-		this.destination = record.topic();
+		this.record = record;
 		String cluster = clusterId.get();
 		setRemoteServiceName("Apache Kafka" + (cluster != null ? ": " + cluster : ""));
 	}
 
+	/**
+	 * Return the template's bean name.
+	 * @return the name.
+	 */
 	public String getBeanName() {
 		return this.beanName;
 	}
@@ -54,7 +58,16 @@ public String getBeanName() {
 	 * @return the topic.
 	 */
 	public String getDestination() {
-		return this.destination;
+		return this.record.topic();
+	}
+
+	/**
+	 * Return the producer record.
+	 * @return the record the record.
+	 * @since 3.0.6
+	 */
+	public ProducerRecord<?, ?> getRecord() {
+		return this.record;
 	}
 
 }