From c8940851ed2cbc2b688f456de02ed9d4e6ecde8d Mon Sep 17 00:00:00 2001
From: Eric Firth <eric.firth@datadoghq.com>
Date: Mon, 24 Feb 2025 14:15:48 -0500
Subject: [PATCH] [DSM] Add messaging.destination tag to spans so they show up
 in queue pages (#5239)

---
 packages/datadog-plugin-kafkajs/src/consumer.js    | 5 ++++-
 packages/datadog-plugin-kafkajs/src/producer.js    | 4 +++-
 packages/datadog-plugin-kafkajs/test/index.spec.js | 7 +++++--
 3 files changed, 12 insertions(+), 4 deletions(-)

diff --git a/packages/datadog-plugin-kafkajs/src/consumer.js b/packages/datadog-plugin-kafkajs/src/consumer.js
index ee04c5eb60c..87ee1ad6e31 100644
--- a/packages/datadog-plugin-kafkajs/src/consumer.js
+++ b/packages/datadog-plugin-kafkajs/src/consumer.js
@@ -7,6 +7,8 @@ const ConsumerPlugin = require('../../dd-trace/src/plugins/consumer')
 const afterStartCh = dc.channel('dd-trace:kafkajs:consumer:afterStart')
 const beforeFinishCh = dc.channel('dd-trace:kafkajs:consumer:beforeFinish')
 
+const MESSAGING_DESTINATION_KEY = 'messaging.destination.name'
+
 class KafkajsConsumerPlugin extends ConsumerPlugin {
   static get id () { return 'kafkajs' }
   static get operation () { return 'consume' }
@@ -72,7 +74,8 @@ class KafkajsConsumerPlugin extends ConsumerPlugin {
         component: 'kafkajs',
         'kafka.topic': topic,
         'kafka.message.offset': message.offset,
-        'kafka.cluster_id': clusterId
+        'kafka.cluster_id': clusterId,
+        [MESSAGING_DESTINATION_KEY]: topic
       },
       metrics: {
         'kafka.partition': partition
diff --git a/packages/datadog-plugin-kafkajs/src/producer.js b/packages/datadog-plugin-kafkajs/src/producer.js
index aa12357b4cf..6e710f88d6c 100644
--- a/packages/datadog-plugin-kafkajs/src/producer.js
+++ b/packages/datadog-plugin-kafkajs/src/producer.js
@@ -5,6 +5,7 @@ const { DsmPathwayCodec } = require('../../dd-trace/src/datastreams/pathway')
 const { getMessageSize } = require('../../dd-trace/src/datastreams/processor')
 
 const BOOTSTRAP_SERVERS_KEY = 'messaging.kafka.bootstrap.servers'
+const MESSAGING_DESTINATION_KEY = 'messaging.destination.name'
 
 class KafkajsProducerPlugin extends ProducerPlugin {
   static get id () { return 'kafkajs' }
@@ -72,7 +73,8 @@ class KafkajsProducerPlugin extends ProducerPlugin {
       meta: {
         component: 'kafkajs',
         'kafka.topic': topic,
-        'kafka.cluster_id': clusterId
+        'kafka.cluster_id': clusterId,
+        [MESSAGING_DESTINATION_KEY]: topic
       },
       metrics: {
         'kafka.batch_size': messages.length
diff --git a/packages/datadog-plugin-kafkajs/test/index.spec.js b/packages/datadog-plugin-kafkajs/test/index.spec.js
index f67279bdd9f..84f3b967f4f 100644
--- a/packages/datadog-plugin-kafkajs/test/index.spec.js
+++ b/packages/datadog-plugin-kafkajs/test/index.spec.js
@@ -73,7 +73,9 @@ describe('Plugin', () => {
             const meta = {
               'span.kind': 'producer',
               component: 'kafkajs',
-              'pathway.hash': expectedProducerHash.readBigUInt64BE(0).toString()
+              'pathway.hash': expectedProducerHash.readBigUInt64BE(0).toString(),
+              'messaging.destination.name': 'test-topic',
+              'messaging.kafka.bootstrap.servers': '127.0.0.1:9092'
             }
             if (clusterIdAvailable) meta['kafka.cluster_id'] = testKafkaClusterId
 
@@ -182,7 +184,8 @@ describe('Plugin', () => {
               meta: {
                 'span.kind': 'consumer',
                 component: 'kafkajs',
-                'pathway.hash': expectedConsumerHash.readBigUInt64BE(0).toString()
+                'pathway.hash': expectedConsumerHash.readBigUInt64BE(0).toString(),
+                'messaging.destination.name': 'test-topic'
               },
               resource: testTopic,
               error: 0,