Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Handle null group id in listener observation (#3778)

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

Fixes: #3778

Signed-off-by: cfredri4 <christian.fredriksson.2@volvocars.com>

**Auto-cherry-pick to `3.3.x` & `3.2.x`**
This commit is contained in:
Christian Fredriksson
2025-03-06 02:50:23 +01:00
committed by GitHub
parent bdd1fd3556
commit ccf466606c
2 changed files with 68 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -32,6 +32,7 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
* @author Christian Mergenthaler
* @author Wang Zhiyang
* @author Christian Fredriksson
*
* @since 3.0
*
@@ -224,26 +225,33 @@ public enum KafkaListenerObservation implements ObservationDocumentation {
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 enum KafkaListenerObservation implements ObservationDocumentation {
.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 enum KafkaListenerObservation implements ObservationDocumentation {
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;
}
}

42
spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/KafkaListenerObservationTests.java Normal file
View File

@@ -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);
}
}