Handle null group id in listener observation (#3778)

This change fixes an NPE when group id is null and observation is enabled.

Fixes: #3778

Signed-off-by: cfredri4 <[email protected]>

(cherry picked from commit ccf4666)

Author: cfredri4

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

@@ -32,6 +32,7 @@
  * @author Gary Russell
  * @author Christian Mergenthaler
  * @author Wang Zhiyang
+ * @author Christian Fredriksson
  *
  * @since 3.0
  *
@@ -224,33 +225,45 @@ public static class DefaultKafkaListenerObservationConvention implements KafkaLi
 				new DefaultKafkaListenerObservationConvention();
 
 		@Override
+		@NonNull
 		public KeyValues getLowCardinalityKeyValues(KafkaRecordReceiverContext context) {
-
-			return KeyValues.of(
+			String groupId = context.getGroupId();
+			KeyValues keyValues = KeyValues.of(
 					ListenerLowCardinalityTags.LISTENER_ID.withValue(context.getListenerId()),
 					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())
+					ListenerLowCardinalityTags.MESSAGING_SOURCE_KIND.withValue("topic")
 			);
+
+			if (StringUtils.hasText(groupId)) {
+				keyValues = keyValues
+						.and(ListenerLowCardinalityTags.MESSAGING_CONSUMER_GROUP.withValue(groupId));
+			}
+
+			return keyValues;
 		}
 
 		@Override
 		@NonNull
 		public KeyValues getHighCardinalityKeyValues(KafkaRecordReceiverContext context) {
 			String clientId = context.getClientId();
+			String consumerId = getConsumerId(context.getGroupId(), clientId);
 			KeyValues keyValues = KeyValues.of(
 					ListenerHighCardinalityTags.MESSAGING_PARTITION.withValue(context.getPartition()),
-					ListenerHighCardinalityTags.MESSAGING_OFFSET.withValue(context.getOffset()),
-					ListenerHighCardinalityTags.MESSAGING_CONSUMER_ID.withValue(getConsumerId(context, clientId))
+					ListenerHighCardinalityTags.MESSAGING_OFFSET.withValue(context.getOffset())
 			);
 
 			if (StringUtils.hasText(clientId)) {
 				keyValues = keyValues
 						.and(ListenerHighCardinalityTags.MESSAGING_CLIENT_ID.withValue(clientId));
 			}
 
+			if (StringUtils.hasText(consumerId)) {
+				keyValues = keyValues
+						.and(ListenerHighCardinalityTags.MESSAGING_CONSUMER_ID.withValue(consumerId));
+			}
+
 			return keyValues;
 		}
 
@@ -259,11 +272,14 @@ public String getContextualName(KafkaRecordReceiverContext context) {
 			return context.getSource() + " receive";
 		}
 
-		private static String getConsumerId(KafkaRecordReceiverContext context, @Nullable String clientId) {
-			if (StringUtils.hasText(clientId)) {
-				return context.getGroupId() + " - " + clientId;
+		private static @Nullable String getConsumerId(@Nullable String groupId, @Nullable String clientId) {
+			if (StringUtils.hasText(groupId)) {
+				if (StringUtils.hasText(clientId)) {
+					return groupId + " - " + clientId;
+				}
+				return groupId;
 			}
-			return context.getGroupId();
+			return clientId;
 		}
 
 	}

spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/KafkaListenerObservationTests.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2020-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support.micrometer;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.kafka.support.micrometer.KafkaListenerObservation.DefaultKafkaListenerObservationConvention;
+
+/**
+ * @author Christian Fredriksson
+ */
+public class KafkaListenerObservationTests {
+
+	@Test
+	void lowCardinalityKeyValues() {
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("topic", 1, 2, "key", "value");
+		KafkaRecordReceiverContext context = new KafkaRecordReceiverContext(record, "listener", () -> null);
+		DefaultKafkaListenerObservationConvention.INSTANCE.getLowCardinalityKeyValues(context);
+	}
+
+	@Test
+	void highCardinalityKeyValues() {
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("topic", 1, 2, "key", "value");
+		KafkaRecordReceiverContext context = new KafkaRecordReceiverContext(record, "listener", () -> null);
+		DefaultKafkaListenerObservationConvention.INSTANCE.getHighCardinalityKeyValues(context);
+	}
+}