feat(kafka): add MessageBroker/Kafka/Cluster/{id}/Produce|Consume/{topic} metrics

lib/feature_flags.js

@@ -13,6 +13,7 @@ exports.prerelease = {
   reverse_naming_rules: false,
   unresolved_promise_cleanup: true,
   kafkajs_instrumentation: false,
+  kafka_cluster_metrics: false,
   undici_error_tracking: true
 }
 

lib/subscribers/kafkajs/client-constructor.js

@@ -14,6 +14,8 @@ const { kafkaCtx } = require('#agentlib/symbols.js')
 const recordDataMetrics = require('./utils/record-data-metrics.js')
 const recordLinkingMetrics = require('./utils/record-linking-metrics.js')
 const recordMethodMetric = require('./utils/record-method-metric.js')
+const recordClusterProduceMetrics = require('./utils/record-cluster-produce-metrics.js')
+const readClusterId = require('./utils/read-cluster-id.js')
 
 const CONSUMER_METHODS = [
   'commitOffsets',
@@ -76,15 +78,19 @@ module.exports = class ConstructorSubscriber extends Subscriber {
   end(data, ctx) {
     const self = this
     const { arguments: args, self: client } = data
-    client[kafkaCtx] = { brokers: args[0].brokers ?? ['none'] }
+    const brokers = args[0]?.brokers
+    client[kafkaCtx] = {
+      brokers: typeof brokers === 'function' ? [] : (brokers ?? ['none'])
+    }
 
     const origConsumer = client.consumer
     client.consumer = function nrConsumer(...args) {
       const consumer = origConsumer.apply(client, args)
       consumer[kafkaCtx] = client[kafkaCtx]
+      consumer[kafkaCtx].clientId = args[0]?.clientId ?? client[kafkaCtx]?.clientId ?? null
 
       consumer.on(consumer.events.REQUEST, function nrListener(data) {
-        consumer[kafkaCtx].clientId = data?.payload?.clientId
+        consumer[kafkaCtx].clientId ??= data?.payload?.clientId
       })
       for (const method of CONSUMER_METHODS) {
         self.#wrapConsumerMethod(consumer, method)
@@ -174,6 +180,8 @@ module.exports = class ConstructorSubscriber extends Subscriber {
           'Not recording consumer function %s, not in a transaction',
           methodName
         )
+        const [data] = args
+        self.#refreshAndRecordProduceMetrics(instance, batch, data)
         return orig.apply(instance, args)
       }
 
@@ -196,24 +204,38 @@ module.exports = class ConstructorSubscriber extends Subscriber {
         brokers: instance[kafkaCtx].brokers,
         topic
       })
+      self.#insertProduceDTHeaders(ctx, batch, data)
+      self.#refreshAndRecordProduceMetrics(instance, batch, data)
 
-      if (batch === false) {
-        for (const msg of data.messages) {
+      return self.agent.tracer.runInContext({ handler: orig, context: ctx, full: true, thisArg: instance, args })
+    }
+  }
+
+  #insertProduceDTHeaders(ctx, batch, data) {
+    if (batch === false) {
+      for (const msg of data.messages) {
+        const headers = msg.headers ?? {}
+        this.insertDTHeaders({ ctx, headers, useMqNames: true })
+        msg.headers = headers
+      }
+    } else {
+      for (const topicMessage of data.topicMessages) {
+        for (const msg of topicMessage.messages) {
           const headers = msg.headers ?? {}
-          self.insertDTHeaders({ ctx, headers, useMqNames: true })
+          this.insertDTHeaders({ ctx, headers, useMqNames: true })
           msg.headers = headers
         }
-      } else {
-        for (const topicMessage of data.topicMessages) {
-          for (const msg of topicMessage.messages) {
-            const headers = msg.headers ?? {}
-            self.insertDTHeaders({ ctx, headers, useMqNames: true })
-            msg.headers = headers
-          }
-        }
       }
+    }
+  }
 
-      return self.agent.tracer.runInContext({ handler: orig, context: ctx, full: true, thisArg: instance, args })
+  #refreshAndRecordProduceMetrics(instance, batch, data) {
+    if (!this.agent.config.feature_flag.kafka_cluster_metrics) {
+      return
+    }
+    const clusterId = readClusterId(instance)
+    if (clusterId) {
+      recordClusterProduceMetrics(this.agent.metrics, clusterId, batch, data)
     }
   }
 
@@ -285,6 +307,8 @@ module.exports = class ConstructorSubscriber extends Subscriber {
         recordDataMetrics({
           tx: ctx.transaction,
           kafkaCtx: instance[kafkaCtx],
+          clusterId: self.agent.config.feature_flag.kafka_cluster_metrics ? readClusterId(instance) : undefined,
+          agentMetrics: self.agent.metrics,
           data
         })
 
@@ -335,12 +359,24 @@ module.exports = class ConstructorSubscriber extends Subscriber {
       const eachBatch = args[0].eachBatch
       args[0].eachBatch = function nrWrappedEachBatch() {
         recordMethodMetric({ agent: self.agent, name: 'eachBatch' })
+        const { batch } = arguments[0]
         recordLinkingMetrics({
           brokers: instance[kafkaCtx].brokers,
           agent: self.agent,
-          topic: arguments[0].batch.topic,
+          topic: batch.topic,
           producer: false
         })
+        // Emit one cluster Consume metric per message in the batch (opt-in only).
+        if (self.agent.config.feature_flag.kafka_cluster_metrics) {
+          const clusterId = readClusterId(instance)
+          if (clusterId) {
+            self.agent.metrics
+              .getOrCreateMetric(
+                `MessageBroker/Kafka/Cluster/${clusterId}/Consume/${batch.topic}`
+              )
+              .incrementCallCount(batch.messages.length)
+          }
+        }
         return eachBatch.apply(instance, arguments)
       }
       return self.agent.tracer.runInContext({ handler: orig, context: ctx, full: true, thisArg: instance, args })

lib/subscribers/kafkajs/config.js

@@ -21,5 +21,37 @@ module.exports = {
         className: 'Client'
       }
     }]
+  }, {
+    // kafkajs builds a brand new internal `Cluster` on every call to
+    // `.producer()` — it isn't reachable from the top-level client, so we
+    // have to capture it right where kafkajs itself creates it.
+    path: './kafkajs/producer-cluster-capture.js',
+    instrumentations: [{
+      module: {
+        name: modName,
+        filePath: 'src/producer/index.js',
+        versionRange: '>=2.0.0'
+      },
+      channelName: 'nr_producerClusterCapture',
+      functionQuery: {
+        expressionName: 'exports',
+        kind: 'Sync'
+      }
+    }]
+  }, {
+    // Same as above, for `.consumer()`.
+    path: './kafkajs/consumer-cluster-capture.js',
+    instrumentations: [{
+      module: {
+        name: modName,
+        filePath: 'src/consumer/index.js',
+        versionRange: '>=2.0.0'
+      },
+      channelName: 'nr_consumerClusterCapture',
+      functionQuery: {
+        expressionName: 'exports',
+        kind: 'Sync'
+      }
+    }]
   }]
 }

lib/subscribers/kafkajs/consumer-cluster-capture.js

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2026 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const Subscriber = require('../base.js')
+const { kafkaCluster } = require('#agentlib/symbols.js')
+
+/**
+ * Same capture as `producer-cluster-capture.js`, but for the internal
+ * consumer factory (`kafkajs/src/consumer/index.js`) — kafkajs also
+ * constructs a brand new `Cluster` instance on every call to
+ * `Kafka.prototype.consumer()`.
+ *
+ * @type {ConsumerClusterCaptureSubscriber}
+ */
+module.exports = class ConsumerClusterCaptureSubscriber extends Subscriber {
+  constructor({ agent, logger }) {
+    super({ agent, logger, channelName: 'nr_consumerClusterCapture', packageName: 'kafkajs' })
+    this.requireActiveTx = false
+    this.events = ['end']
+  }
+
+  get enabled() {
+    if (this.agent.config.feature_flag.kafkajs_instrumentation === false) {
+      return false
+    }
+
+    return super.enabled
+  }
+
+  end(data, ctx) {
+    const cluster = data?.arguments?.[0]?.cluster
+    if (cluster && data.result) {
+      data.result[kafkaCluster] = cluster
+    }
+    return ctx
+  }
+}

lib/subscribers/kafkajs/producer-cluster-capture.js

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2026 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const Subscriber = require('../base.js')
+const { kafkaCluster } = require('#agentlib/symbols.js')
+
+/**
+ * kafkajs constructs a brand new internal `Cluster` instance on every call
+ * to `Kafka.prototype.producer()` — it is not reachable from the top-level
+ * `Kafka` client. This subscriber intercepts the internal producer factory
+ * (`kafkajs/src/producer/index.js`) at the point where kafkajs itself
+ * passes the freshly created `cluster` in, and stashes a reference to it on
+ * the returned producer instance so `read-cluster-id.js` can read the
+ * cluster id straight off it later — no admin connection, no cache needed.
+ *
+ * @type {ProducerClusterCaptureSubscriber}
+ */
+module.exports = class ProducerClusterCaptureSubscriber extends Subscriber {
+  constructor({ agent, logger }) {
+    super({ agent, logger, channelName: 'nr_producerClusterCapture', packageName: 'kafkajs' })
+    this.requireActiveTx = false
+    this.events = ['end']
+  }
+
+  get enabled() {
+    if (this.agent.config.feature_flag.kafkajs_instrumentation === false) {
+      return false
+    }
+
+    return super.enabled
+  }
+
+  end(data, ctx) {
+    const cluster = data?.arguments?.[0]?.cluster
+    if (cluster && data.result) {
+      data.result[kafkaCluster] = cluster
+    }
+    return ctx
+  }
+}

lib/subscribers/kafkajs/utils/read-cluster-id.js

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2026 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const { kafkaCluster } = require('#agentlib/symbols.js')
+
+/**
+ * Reads the Kafka cluster id off a producer/consumer instance. The `Cluster`
+ * reference is captured at producer()/consumer() creation time (see
+ * `producer-cluster-capture.js` / `consumer-cluster-capture.js`) and stashed
+ * on the instance via the `kafkaCluster` symbol. kafkajs keeps
+ * `cluster.brokerPool.metadata` current on its own as part of normal
+ * operation, so this is a synchronous, in-memory, best-effort read — no
+ * network call, no cache.
+ *
+ * @param {object} instance Producer or consumer client instance.
+ * @returns {string|undefined} The cluster id, or `undefined` if not yet
+ * available (e.g. a producer's very first send, before kafkajs has fetched
+ * any metadata).
+ */
+module.exports = function readClusterId(instance) {
+  try {
+    const clusterId = instance?.[kafkaCluster]?.brokerPool?.metadata?.clusterId
+    return typeof clusterId === 'string' && clusterId !== '' ? clusterId : undefined
+  } catch {
+    return undefined
+  }
+}

lib/subscribers/kafkajs/utils/record-cluster-produce-metrics.js

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2026 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+/**
+ * Records MessageBroker/Kafka/Cluster cluster-level produce metrics.
+ * For send() records one metric for the single topic.
+ * For sendBatch() records one metric per distinct topic in the batch.
+ *
+ * @param {object} metrics The agent metrics aggregator.
+ * @param {string} clusterId Kafka cluster UUID.
+ * @param {boolean} batch Whether this is a sendBatch call.
+ * @param {object} data The send/sendBatch arguments object.
+ */
+module.exports = function recordClusterProduceMetrics(metrics, clusterId, batch, data) {
+  if (batch === false) {
+    if (!Array.isArray(data?.messages)) {
+      return
+    }
+    metrics
+      .getOrCreateMetric(`MessageBroker/Kafka/Cluster/${clusterId}/Produce/${data.topic}`)
+      .incrementCallCount(data.messages.length)
+  } else {
+    if (!Array.isArray(data?.topicMessages)) {
+      return
+    }
+    for (const topicMessage of data.topicMessages) {
+      if (!Array.isArray(topicMessage?.messages)) {
+        continue
+      }
+      metrics
+        .getOrCreateMetric(
+          `MessageBroker/Kafka/Cluster/${clusterId}/Produce/${topicMessage.topic}`
+        )
+        .incrementCallCount(topicMessage.messages.length)
+    }
+  }
+}

lib/subscribers/kafkajs/utils/record-data-metrics.js

@@ -17,8 +17,10 @@ const { DESTINATIONS } = require('#agentlib/config/attribute-filter.js')
  * @param {object} params.kafkaCtx The local context store we add to the
  * consumer client.
  * @param {Transaction} params.tx The current transaction.
+ * @param {string} [params.clusterId] Kafka cluster id, if resolved and cluster metrics are enabled.
+ * @param {object} params.agentMetrics Agent-level metrics aggregator (for cluster metrics).
  */
-module.exports = function recordDataMetrics({ data, kafkaCtx, tx }) {
+module.exports = function recordDataMetrics({ data, kafkaCtx, tx, clusterId, agentMetrics }) {
   if (!tx) {
     return
   }
@@ -45,4 +47,11 @@ module.exports = function recordDataMetrics({ data, kafkaCtx, tx }) {
       kafkaCtx.clientId
     )
   }
+  if (clusterId && agentMetrics) {
+    agentMetrics
+      .getOrCreateMetric(
+        `MessageBroker/Kafka/Cluster/${clusterId}/Consume/${topic}`
+      )
+      .incrementCallCount()
+  }
 }

lib/symbols.js

@@ -13,6 +13,7 @@ module.exports = {
   databaseName: Symbol('databaseName'),
   disableDT: Symbol('Disable distributed tracing'), // description for backwards compatibility
   executorContext: Symbol('executorContext'),
+  kafkaCluster: Symbol('kafkaCluster'),
   kafkaCtx: Symbol('kafkaCtx'),
   name: Symbol('name'),
   onceExecuted: Symbol('onceExecuted'),

test/unit/feature_flag.test.js

@@ -42,6 +42,7 @@ const used = [
   'aws_bedrock_instrumentation',
   'langchain_instrumentation',
   'kafkajs_instrumentation',
+  'kafka_cluster_metrics',
   'undici_error_tracking'
 ]