* fix review.

* polish `ObservationTests`.
* add @author.
* add doc in `whats-new.adoc`.

Author: Wzy19930507

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -96,3 +96,10 @@ See xref:kafka/connecting.adoc#default-client-id-prefixes[Default client ID pref
 `getListenerContainersMatching(BiPredicate<String, MessageListenerContainer> matcher)` to filter by ID and container properties.
 
 See xref:kafka/receiving-messages/kafkalistener-lifecycle.adoc#retrieving-message-listener-containers[`@KafkaListener` Lifecycle Management's API Docs] for more information.
+
+[[x32-observation]]
+== Enhanced observation by providing more tracing tags
+
+`KafkaTemplateObservation` provides more tracing tags(low cardinality).
+`KafkaListenerObservation` provides a new API to find high cardinality key names and more tracing tags(high or low cardinality).
+See xref:kafka/micrometer.adoc#observation[Micrometer Observation]

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

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.support.micrometer;
 
+import org.springframework.lang.NonNull;
 import org.springframework.util.StringUtils;
 
 import io.micrometer.common.KeyValues;
@@ -28,6 +29,9 @@
  * Spring for Apache Kafka Observation for listeners.
  *
  * @author Gary Russell
+ * @author Christian Mergenthaler
+ * @author Wang Zhiyang
+ *
  * @since 3.0
  *
  */
@@ -44,19 +48,32 @@ public Class<? extends ObservationConvention<? extends Context>> getDefaultConve
 		}
 
 		@Override
+		@NonNull
 		public String getPrefix() {
 			return "spring.kafka.listener";
 		}
 
 		@Override
+		@NonNull
 		public KeyName[] getLowCardinalityKeyNames() {
 			return ListenerLowCardinalityTags.values();
 		}
 
+		@Override
+		@NonNull
+		public KeyName[] getHighCardinalityKeyNames() {
+			return ListenerHighCardinalityTags.values();
+		}
+
 	};
 
 	/**
 	 * Low cardinality tags.
+	 *
+	 * @author Christian Mergenthaler
+	 * @author Wang Zhiyang
+	 *
+	 * @since 3.2
 	 */
 	public enum ListenerLowCardinalityTags implements KeyName {
 
@@ -66,6 +83,7 @@ public enum ListenerLowCardinalityTags implements KeyName {
 		LISTENER_ID {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "spring.kafka.listener.id";
 			}
@@ -78,6 +96,7 @@ public String asString() {
 		MESSAGING_SYSTEM {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "messaging.system";
 			}
@@ -90,30 +109,20 @@ public String asString() {
 		MESSAGING_OPERATION {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "messaging.operation";
 			}
 
 		},
 
-		/**
-		 * Messaging consumer id.
-		 */
-		MESSAGING_CONSUMER_ID {
-
-			@Override
-			public String asString() {
-				return "messaging.consumer.id";
-			}
-
-		},
-
 		/**
 		 * Messaging source name.
 		 */
 		MESSAGING_SOURCE_NAME {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "messaging.source.name";
 			}
@@ -126,42 +135,71 @@ public String asString() {
 		MESSAGING_SOURCE_KIND {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "messaging.source.kind";
 			}
 
 		},
 
 		/**
-		 * Messaging consumer group.
+		 * Messaging the consumer group.
 		 */
 		MESSAGING_CONSUMER_GROUP {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "messaging.kafka.consumer.group";
 			}
 
 		},
 
+	}
+
+	/**
+	 * High cardinality tags.
+	 *
+	 * @author Wang Zhiyang
+	 * @author Christian Mergenthaler
+	 *
+	 * @since 3.2
+	 */
+	public enum ListenerHighCardinalityTags implements KeyName {
+
 		/**
 		 * Messaging client id.
 		 */
 		MESSAGING_CLIENT_ID {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "messaging.kafka.client_id";
 			}
 
 		},
 
+		/**
+		 * Messaging consumer id (consumer group and client id).
+		 */
+		MESSAGING_CONSUMER_ID {
+
+			@Override
+			@NonNull
+			public String asString() {
+				return "messaging.consumer.id";
+			}
+
+		},
+
 		/**
 		 * Messaging partition.
 		 */
 		MESSAGING_PARTITION {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "messaging.kafka.source.partition";
 			}
@@ -174,18 +212,22 @@ public String asString() {
 		MESSAGING_OFFSET {
 
 			@Override
+			@NonNull
 			public String asString() {
 				return "messaging.kafka.message.offset";
 			}
 
-		}
+		},
 
 	}
 
 	/**
 	 * Default {@link KafkaListenerObservationConvention} for Kafka listener key values.
 	 *
 	 * @author Gary Russell
+	 * @author Christian Mergenthaler
+	 * @author Wang Zhiyang
+	 *
 	 * @since 3.0
 	 *
 	 */
@@ -199,30 +241,31 @@ public static class DefaultKafkaListenerObservationConvention implements KafkaLi
 
 		@Override
 		public KeyValues getLowCardinalityKeyValues(KafkaRecordReceiverContext context) {
-			KeyValues keyValues = KeyValues.of(
+
+			return KeyValues.of(
 					ListenerLowCardinalityTags.LISTENER_ID.withValue(context.getListenerId()),
-					ListenerLowCardinalityTags.MESSAGING_CONSUMER_ID.withValue(getConsumerId(context)),
 					ListenerLowCardinalityTags.MESSAGING_SYSTEM.withValue("kafka"),
 					ListenerLowCardinalityTags.MESSAGING_OPERATION.withValue("receive"),
 					ListenerLowCardinalityTags.MESSAGING_SOURCE_NAME.withValue(context.getSource()),
 					ListenerLowCardinalityTags.MESSAGING_SOURCE_KIND.withValue("topic"),
 					ListenerLowCardinalityTags.MESSAGING_CONSUMER_GROUP.withValue(context.getGroupId())
 			);
-
-			if (StringUtils.hasText(context.getClientId())) {
-				keyValues = keyValues.and(ListenerLowCardinalityTags.MESSAGING_CLIENT_ID.withValue(context.getClientId()));
-			}
-
-			return keyValues;
 		}
 
 		@Override
+		@NonNull
 		public KeyValues getHighCardinalityKeyValues(KafkaRecordReceiverContext context) {
 			KeyValues keyValues = KeyValues.of(
-					ListenerLowCardinalityTags.MESSAGING_PARTITION.withValue(context.getPartition()),
-					ListenerLowCardinalityTags.MESSAGING_OFFSET.withValue(context.getOffset())
+					ListenerHighCardinalityTags.MESSAGING_PARTITION.withValue(context.getPartition()),
+					ListenerHighCardinalityTags.MESSAGING_OFFSET.withValue(context.getOffset())
 			);
 
+			if (StringUtils.hasText(context.getClientId())) {
+				keyValues = keyValues
+						.and(ListenerHighCardinalityTags.MESSAGING_CLIENT_ID.withValue(context.getClientId()))
+						.and(ListenerHighCardinalityTags.MESSAGING_CONSUMER_ID.withValue(getConsumerId(context)));
+			}
+
 			return keyValues;
 		}
 
@@ -231,11 +274,6 @@ public String getContextualName(KafkaRecordReceiverContext context) {
 			return context.getSource() + " receive";
 		}
 
-		@Override
-		public String getName() {
-			return "spring.kafka.listener";
-		}
-
 		private String getConsumerId(KafkaRecordReceiverContext context) {
 			if (StringUtils.hasText(context.getClientId())) {
 				return context.getGroupId() + " - " + context.getClientId();

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

@@ -29,13 +29,17 @@
  *
  * @author Gary Russell
  * @author Christian Mergenthaler
+ * @author Wang Zhiyang
+ *
  * @since 3.0
  *
  */
 public class KafkaRecordReceiverContext extends ReceiverContext<ConsumerRecord<?, ?>> {
 
 	private final String listenerId;
+
 	private final String clientId;
+
 	private final String groupId;
 
 	private final ConsumerRecord<?, ?> record;
@@ -84,7 +88,7 @@ public String getSource() {
 
 	/**
 	 * Return the consumer record.
-	 * @return the record the record.
+	 * @return the record.
 	 * @since 3.0.6
 	 */
 	public ConsumerRecord<?, ?> getRecord() {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -28,6 +28,8 @@
  *
  * @author Gary Russell
  * @author Christian Mergenthaler
+ * @author Wang Zhiyang
+ *
  * @since 3.0
  *
  */
@@ -65,10 +67,11 @@ public String getDestination() {
 
 	/**
 	 * Return the producer record.
-	 * @return the record the record.
+	 * @return the record.
 	 * @since 3.0.6
 	 */
 	public ProducerRecord<?, ?> getRecord() {
 		return this.record;
 	}
+
 }