avoid calculate kafka message size when dsm is not enabled (#6179)

DataDog · Nov 9, 2023 · 2e6162b · 2e6162b
1 parent 80b242d
commit 2e6162b
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 5 deletions.
diff --git a/...ients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java b/...ients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java
@@ -90,10 +90,11 @@ protected void startNewRecordSpan(ConsumerRecord<?, ?> val) {
           sortedTags.put(GROUP_TAG, group);
           sortedTags.put(TOPIC_TAG, val.topic());
           sortedTags.put(TYPE_TAG, "kafka");
-
+          final long payloadSize =
+              span.traceConfig().isDataStreamsEnabled() ? computePayloadSizeBytes(val) : 0;
           AgentTracer.get()
               .getDataStreamsMonitoring()
-              .setCheckpoint(span, sortedTags, val.timestamp(), computePayloadSizeBytes(val));
+              .setCheckpoint(span, sortedTags, val.timestamp(), payloadSize);
         } else {
           span = startSpan(operationName, null);
         }

diff --git a/...n/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/Utils.java b/...n/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/Utils.java
@@ -6,7 +6,7 @@
 import org.apache.kafka.common.header.Headers;
 
 public final class Utils {
-  private Utils() {} // prevent instanciation
+  private Utils() {} // prevent instantiation
 
   // this method is used in kafka-clients and kafka-streams instrumentations
   public static long computePayloadSizeBytes(ConsumerRecord<?, ?> val) {

diff --git a/...main/java/datadog/trace/instrumentation/kafka_streams/KafkaStreamTaskInstrumentation.java b/...main/java/datadog/trace/instrumentation/kafka_streams/KafkaStreamTaskInstrumentation.java
@@ -245,10 +245,11 @@ public static void start(
         }
         sortedTags.put(TOPIC_TAG, record.topic());
         sortedTags.put(TYPE_TAG, "kafka");
+        final long payloadSize =
+            span.traceConfig().isDataStreamsEnabled() ? computePayloadSizeBytes(record.value) : 0;
         AgentTracer.get()
             .getDataStreamsMonitoring()
-            .setCheckpoint(
-                span, sortedTags, record.timestamp, computePayloadSizeBytes(record.value));
+            .setCheckpoint(span, sortedTags, record.timestamp, payloadSize);
       } else {
         span = startSpan(KAFKA_CONSUME, null);
       }