opentelemetry-instrumentation-confluent-kafka

OpenTelemetry auto-instrumentation for @confluentinc/kafka-javascript.

Automatically creates traces for Kafka produce and consume operations and records metrics for message throughput, latency, and consumer lag — all with zero changes to your application code.

Supports both API styles exposed by @confluentinc/kafka-javascript:

• NodeRdKafkaInstrumentation — callback-based (Producer / KafkaConsumer)
• KafkaJSInstrumentation — promise-based KafkaJS-compatible API (KafkaJS.Kafka)

Installation

npm install opentelemetry-instrumentation-confluent-kafka

Peer dependencies — make sure these are present in your project:

npm install @opentelemetry/api @opentelemetry/instrumentation @confluentinc/kafka-javascript

Usage

Register the instrumentation(s) with the OpenTelemetry SDK before your application starts:

Callback API (NodeRdKafkaInstrumentation)

import { NodeSDK } from "@opentelemetry/sdk-node";
import { NodeRdKafkaInstrumentation } from "opentelemetry-instrumentation-confluent-kafka";

const sdk = new NodeSDK({
  instrumentations: [new NodeRdKafkaInstrumentation()],
});

sdk.start();

KafkaJS-compatible API (KafkaJSInstrumentation)

import { NodeSDK } from "@opentelemetry/sdk-node";
import { KafkaJSInstrumentation } from "opentelemetry-instrumentation-confluent-kafka";

const sdk = new NodeSDK({
  instrumentations: [new KafkaJSInstrumentation()],
});

sdk.start();

Your existing producer and consumer code requires no changes.

What Gets Instrumented

NodeRdKafkaInstrumentation — callback API

Producer (Producer)

| Event | Span kind | Span created when | |---|---|---| | producer.produce(...) | PRODUCER | Call to produce() — trace context is injected into message headers and opaque | | producer.on("delivery-report", ...) | PRODUCER | Delivery confirmation received from broker |

Consumer (KafkaConsumer)

| Event | Span kind | Span created when | |---|---|---| | consumer.on("data", ...) | CONSUMER | Message received — trace context is extracted from message headers |

KafkaJSInstrumentation — KafkaJS-compatible API

Producer (KafkaJS.Producer)

| Method | Span kind | Span created when | |---|---|---| | producer.send(...) | PRODUCER | One span per message — trace context is injected into message headers |

sendBatch is not patched separately because its implementation delegates to send() internally, which is already instrumented.

Consumer (KafkaJS.Consumer)

| Callback | Span kind | Span created when | |---|---|---| | consumer.run({ eachMessage }) | CONSUMER | One span per message — trace context extracted from message headers | | consumer.run({ eachBatch }) | CONSUMER | One span per message in the batch — trace context extracted per message |

Metrics (both instrumentations)

| Metric | Type | Unit | Description | |---|---|---|---| | messaging.kafka.consumer.lag | Histogram | ms | Date.now() - message.timestamp — time between produce and consume | | messaging.publish.messages | Counter | {message} | Number of messages produced | | messaging.receive.messages | Counter | {message} | Number of messages consumed | | messaging.publish.duration | Histogram | ms | Duration of producer send operations | | messaging.process.duration | Histogram | ms | Duration of consumer message processing |

Common dimensions on all metrics: messaging.system, messaging.destination.name (topic), messaging.operation.type

messaging.kafka.consumer.lag, messaging.receive.messages, and messaging.process.duration also include messaging.destination.partition_id.

Span Attributes

All spans include the following OpenTelemetry semantic convention attributes:

| Attribute | Value | |---|---| | messaging.system | kafka | | messaging.destination.name | topic name | | messaging.operation.type | send / process / receive |

Consumer spans additionally include:

| Attribute | Value | |---|---| | messaging.destination.partition_id | partition number | | messaging.kafka.offset | message offset |

Delivery-report spans (NodeRdKafkaInstrumentation) additionally include:

| Attribute | Value | |---|---| | messaging.destination.partition_id | partition number | | messaging.kafka.offset | committed offset | | messaging.kafka.timestamp | broker timestamp |

Configuration

Both classes accept the same set of options.

NodeRdKafkaInstrumentation

import { NodeRdKafkaInstrumentation } from "opentelemetry-instrumentation-confluent-kafka";

new NodeRdKafkaInstrumentation({
  /** Called on every produce — add custom attributes to the producer span. */
  producerHook: (span, topic, message) => {
    span.setAttribute("app.message.key", String(message.key));
  },

  /** Called on every consumed message — add custom attributes to the consumer span. */
  consumerHook: (span, topic, message) => {
    span.setAttribute("app.consumer.group", "my-group");
  },

  /** Add the library version as a span attribute under this key. */
  moduleVersionAttributeName: "kafka.client.version",
});

KafkaJSInstrumentation

import { KafkaJSInstrumentation } from "opentelemetry-instrumentation-confluent-kafka";

new KafkaJSInstrumentation({
  /** Called on every producer.send() message — add custom attributes to the producer span. */
  producerHook: (span, topic, message) => {
    span.setAttribute("app.message.key", String(message.key));
  },

  /** Called on every eachMessage / eachBatch message — add custom attributes to the consumer span. */
  consumerHook: (span, topic, message) => {
    span.setAttribute("app.consumer.group", "my-group");
  },

  /** Add the library version as a span attribute under this key. */
  moduleVersionAttributeName: "kafka.client.version",
});

Configuration options

| Option | Type | Description | |---|---|---| | producerHook | (span, topic, message) => void | Hook called for every produced message. Use to enrich the producer span with custom attributes. | | consumerHook | (span, topic, message) => void | Hook called for every consumed message. Use to enrich the consumer span with custom attributes. | | moduleVersionAttributeName | string | If set, the resolved @confluentinc/kafka-javascript version is added to every span under this attribute key. |

Customising the Span Name

In exporters such as Azure Application Insights, the span name becomes operation_Name. Use the hooks to override it:

new NodeRdKafkaInstrumentation({
  producerHook: (span, topic) => span.updateName(`${topic} send`),
  consumerHook: (span, topic) => span.updateName(`${topic} process`),
});

Alternatively, use a custom SpanProcessor:

import { SpanProcessor, ReadableSpan, Span } from "@opentelemetry/sdk-trace-base";

class KafkaOperationNameProcessor implements SpanProcessor {
  onStart(span: Span): void {
    const topic = span.attributes["messaging.destination.name"];
    const op = span.attributes["messaging.operation.type"];
    if (topic && op) span.updateName(`${topic} ${op}`);
  }
  onEnd(_span: ReadableSpan): void {}
  shutdown() { return Promise.resolve(); }
  forceFlush() { return Promise.resolve(); }
}

Trace Propagation

Trace context is propagated using the configured OpenTelemetry propagator (W3C TraceContext by default).

NodeRdKafkaInstrumentation

• Producer — context is injected into both message headers and the opaque field.
• Consumer — context is extracted from the traceparent message header.
• Delivery report — context is restored from opaque to link the delivery span back to the original produce span.

KafkaJSInstrumentation

• Producer — context is injected into message.headers for each message.
• Consumer — context is extracted from message.headers for each message.

Exported Symbols

import {
  // Instrumentation classes
  NodeRdKafkaInstrumentation,
  KafkaJSInstrumentation,

  // Config interfaces
  NodeRdKafkaInstrumentationConfig,
  KafkaJSInstrumentationConfig,

  // Hook function types
  KafkaProducerCustomAttributeFunction,
  KafkaConsumerCustomAttributeFunction,
  KafkaJSProducerCustomAttributeFunction,
  KafkaJSConsumerCustomAttributeFunction,

  // Utility
  bufferTextMapGetter,  // TextMapGetter that handles Buffer-valued headers
} from "opentelemetry-instrumentation-confluent-kafka";

License

ISC