From bdf4dec00b0e33b2fc8464786e4e5ed7b01e1b60 Mon Sep 17 00:00:00 2001
From: Andrea Marziali <andrea.marziali@datadoghq.com>
Date: Fri, 14 Jun 2024 11:57:44 +0200
Subject: [PATCH] Fix NPE when kafka consumer info is not available

---
 .../kafka_clients/ConsumerCoordinatorInstrumentation.java   | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java b/dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java
index 01d92cfc810..5a299a30bda 100644
--- a/dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java
+++ b/dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java
@@ -65,7 +65,7 @@ public static void trackCommitOffset(
         @Advice.This ConsumerCoordinator coordinator,
         @Advice.Return RequestFuture<Void> requestFuture,
         @Advice.Argument(0) final Map<TopicPartition, OffsetAndMetadata> offsets) {
-      if (requestFuture.failed()) {
+      if (requestFuture == null || requestFuture.failed()) {
         return;
       }
       if (offsets == null) {
@@ -75,6 +75,10 @@ public static void trackCommitOffset(
           InstrumentationContext.get(ConsumerCoordinator.class, KafkaConsumerInfo.class)
               .get(coordinator);
 
+      if (kafkaConsumerInfo == null) {
+        return;
+      }
+
       String consumerGroup = kafkaConsumerInfo.getConsumerGroup();
       Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata();
       String clusterId = null;