From 845d3e690bed0915c452aa1328b49fc7b3d6760b Mon Sep 17 00:00:00 2001 From: cfredri4 Date: Wed, 5 Mar 2025 12:18:18 +0100 Subject: [PATCH] Handle null group id in listener observation This change fixes an NPE when group id is null and observation is enabled. Signed-off-by: cfredri4 --- .../micrometer/KafkaListenerObservation.java | 36 +++++++++++----- .../KafkaListenerObservationTests.java | 42 +++++++++++++++++++ 2 files changed, 68 insertions(+), 10 deletions(-) create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/KafkaListenerObservationTests.java diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaListenerObservation.java b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaListenerObservation.java index d2d77d756b..ec9dcf4b4b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaListenerObservation.java +++ b/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,26 +225,33 @@ 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)) { @@ -251,6 +259,11 @@ public KeyValues getHighCardinalityKeyValues(KafkaRecordReceiverContext context) .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 @Nullable 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; } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/KafkaListenerObservationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/KafkaListenerObservationTests.java new file mode 100644 index 0000000000..a3ccdecbe8 --- /dev/null +++ b/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 record = new ConsumerRecord<>("topic", 1, 2, "key", "value"); + KafkaRecordReceiverContext context = new KafkaRecordReceiverContext(record, "listener", () -> null); + DefaultKafkaListenerObservationConvention.INSTANCE.getLowCardinalityKeyValues(context); + } + + @Test + void highCardinalityKeyValues() { + ConsumerRecord record = new ConsumerRecord<>("topic", 1, 2, "key", "value"); + KafkaRecordReceiverContext context = new KafkaRecordReceiverContext(record, "listener", () -> null); + DefaultKafkaListenerObservationConvention.INSTANCE.getHighCardinalityKeyValues(context); + } +}