NPE fix for null keys in consumer records (#219)

Tinkoff · Jul 26, 2023 · f11639e · f11639e
1 parent 6e1c16f
commit f11639e
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...ru/tinkoff/kora/opentelemetry/module/kafka/consumer/OpentelemetryKafkaConsumerTracer.java b/...ru/tinkoff/kora/opentelemetry/module/kafka/consumer/OpentelemetryKafkaConsumerTracer.java
@@ -18,6 +18,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 
 public class OpentelemetryKafkaConsumerTracer implements KafkaConsumerTracer {
     private final Tracer tracer;
@@ -83,9 +84,11 @@ public KafkaConsumerRecordSpan get(ConsumerRecord<?, ?> record) {
                 .setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "kafka")
                 .setAttribute(SemanticAttributes.MESSAGING_SOURCE_NAME, record.topic())
                 .setAttribute(SemanticAttributes.MESSAGING_SOURCE_KIND, "topic")
-                .setAttribute(SemanticAttributes.MESSAGING_KAFKA_MESSAGE_KEY, record.key().toString())
                 .setAttribute(SemanticAttributes.MESSAGING_KAFKA_SOURCE_PARTITION, (long) record.partition())
                 .setAttribute(SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET, record.offset());
+            try {
+                recordSpanBuilder.setAttribute(SemanticAttributes.MESSAGING_KAFKA_MESSAGE_KEY, Objects.toString(record.key()));
+            } catch (Exception ignore) {}
             var recordSpan = recordSpanBuilder.startSpan();
             OpentelemetryContext.set(Context.current(), this.rootCtx.add(recordSpan));