---
title: Java agent: Instrument Kafka message queues
source: https://docs.newrelic.com/docs/apm/agents/java-agent/instrumentation/java-agent-instrument-kafka-message-queues
---

The New Relic Java agent automatically collects data from [Kafka](https://kafka.apache.org/documentation/)'s Java clients library. Because Kafka is a high-performance messaging system that generates a lot of data, you can customize the agent for your app's specific throughput and use cases.

This document explains how to collect and view these Kafka data types:

-   [Kafka metrics](#view-kafka-metrics)
-   [Kafka events](#collect-kafka-events)
-   [Kafka Streams transactions](#collect-kafka-streams-transactions)
-   [Kafka distributed traces](#collect-kafka-distributed-traces)

> #### 💡 TIP
>
> We also have a Kafka integration. For details on that, see [Kafka monitoring integration](https://docs.newrelic.com/docs/integrations/host-integrations/host-integrations-list/kafka-monitoring-integration).

## Requirements [#requirements]

Kafka clients instrumentation is available in Java agent versions 4.12.0 or higher. Kafka streams instrumentation is available in Java Agent versions 8.1.0 or higher. To see all supported Kafka libraries, check the [Java compatibility and requirements page](https://docs.newrelic.com/docs/agents/java-agent/getting-started/compatibility-requirements-java-agent). Note that Kafka Streams runs on top of Kafka clients so all of the instrumentation that applies to Kafka clients also applies to Streams.

## View Kafka metrics

After [installation](https://docs.newrelic.com/docs/agents/java-agent/installation/install-java-agent), the Java agent automatically reports rich Kafka metrics with information about messaging rates, latency, lag, and more. The agent collects all [Kafka consumer and producer metrics](https://kafka.apache.org/documentation/#monitoring) (but not connect or stream metrics).

To view these metrics, create a custom dashboard:

1.  Go to the [New Relic metric explorer](https://docs.newrelic.com/docs/insights/use-insights-ui/explore-data/metric-explorer-search-chart-metrics-sent-new-relic-agents).

2.  Use the metric explorer to locate your metrics. Here are some folders where you can find metrics:

    -   Kafka metrics:

        ```
        MessageBroker/Kafka/Internal/KafkaMetricName
        ```

        For example, `request-rate` metric:

        ```
        MessageBroker/Kafka/Internal/consumer-metrics/request-rate
        ```

    -   Kafka Streams:

        ```
        Kafka/Streams/KafkaStreamsMetricName
        ```

        For example, `poll-latency-avg` metric:

        ```
        Kafka/Streams/stream-thread-metrics/poll-latency-avg
        ```

    -   Kafka Connect:

        ```
        Kafka/Connect/KafkaConnectMetricName
        ```

        For example, `connector-count` metric:

        ```
        Kafka/Connect/connect-worker-metrics/connector-count
        ```

3.  Add the metrics you want to monitor to a dashboard by clicking **Add to dashboard**.

> #### 💡 TIP
>
> For a full list of Kafka consumer, producer, and streams metrics, see the [Kafka docs](https://kafka.apache.org/documentation/#remote_jmx). The metrics in those docs are searchable via JMX. Keep in mind not every metric mentioned in the docs will be exported into New Relic. This could be due to one of these reasons:
>
> -   The metric is not actually generated by Kafka clients or Kafka Streams. This may be due to using an older version of clients or Streams or based on how you set up and use your Kafka libraries.
> -   The metric is not numeric or its value is `NaN`. New Relic only accepts metrics with a numeric value.

## View Kafka event collection [#collect-kafka-events]

You can configure the agent to collect event data instead of metric timeslice data (for the difference between metric timeslice and event data, see [data collection](https://docs.newrelic.com/docs/using-new-relic/metrics/analyze-your-metrics/data-collection-metric-timeslice-event-data#overview)). This allows you to use [NRQL](https://docs.newrelic.com/docs/insights/nrql-new-relic-query-language/using-nrql/introduction-nrql) to filter and facet the default Kafka metrics. When enabled, the agent collects one Kafka event every 30 seconds. This event contains all of the the data from [Kafka consumer and produce metrics](https://kafka.apache.org/documentation/#monitoring) captured since the previous event.

If you are using Kafka Streams, the agent generates a seperate event that contains all of the data from [Kafka stream metrics](https://kafka.apache.org/documentation/#remote_jmx) captured since the previous event. The event is also collected every 30 seconds.

> #### ⚠️ IMPORTANT
>
> The agent records up to 2000 events per [harvest cycle](https://docs.newrelic.com/docs/using-new-relic/welcome-new-relic/getting-started/glossary#harvest-cycle), though you can change this value with [`max_samples_stored`](https://docs.newrelic.com/docs/agents/java-agent/configuration/java-agent-configuration-config-file#ae-max_samples_stored). Kafka event data is included in this pool. If you use the `recordCustomEvent()` API call to send [custom events](https://docs.newrelic.com/docs/insights/insights-data-sources/custom-data/insert-custom-events-new-relic-apm-agents) to New Relic and you send more than 2000 events, the agent will discard some Kafka or custom events.

To enable Kafka event collection:

1.  Add the `kafka.metrics.as_events.enabled` element to your `newrelic.yml` config file:

    ```yml
    kafka.metrics.as_events.enabled: true
    ```
2.  Restart your JVM.
3.  Use the [event explorer](https://docs.newrelic.com/docs/insights/use-insights-ui/explore-data/event-explorer-query-chart-your-event-data) to view your Kafka events, located in the `KafkaMetrics` event type. Or, use [NRQL](https://docs.newrelic.com/docs/insights/nrql-new-relic-query-language/using-nrql/introduction-nrql) to query your events directly. For example:

    ```sql
    SELECT average('producer-metrics.record-send-rate') FROM KafkaMetrics SINCE 30 minutes ago TIMESERIES
    ```

    If you are querying Kafka Streams metrics, use the `KafkaStreamsMetrics` event type to access streams-specific metrics.

> #### ⚠️ IMPORTANT
>
> Keep in mind that the limitations on what kind of Kafka metrics you can send to New Relic as timeslice metrics also applies to events. That is, non-numeric and NaN metrics are not included as event attributes.

## Enable Kafka node metrics [#kafka-node-metrics]

> #### ⚠️ IMPORTANT
>
> Kafka Clients Node Metrics instrumentation may not load for users of the `kafka-clients` library versions 3.7.0-3.8.x and Java Agent versions 8.15.0-8.17.0.
>
> To enable Kafka Clients Node Metrics for `kafka-clients` 3.7.0-3.8.x, please upgrade your Java Agent to version 8.18.0 or higher.

There is an alternative instrumentation module for Kafka clients that will provide more granularity for Kafka metrics. This instrumentation module is available since agent 8.6.0 and is disabled by default.

To enabled this instrumentation module, you have to disable the existing instrumentation module and enable the new one by adding the following to your `newrelic.yml` config file:

```yml
class_transformer:
  kafka-clients-metrics:
    enabled: false
  kafka-clients-node-metrics:
    enabled: true
```

## Kafka config events [#kafka-config]

The `kafka-clients-config` instrumentation module will periodically send events with the contents of your Kafka client configuration. This module is available since agent 8.6.0 and is disabled by default.

To enable `kafka-clients-config` add the following to your `newrelic.yml` config file:

```yml
class_transformer:
  kafka-clients-config:
    enabled: true
```

## Kafka Streams transactions [#collect-kafka-streams-transactions]

If you're using Kafka Streams, by default we do not enable transactions. This is to prevent unnecessary overhead because Kafka applications tend to have high throughput.

Unlike JMS transactions, Kafka Streams transactions are not processed per record. Instead, a transaction begins when a Kafka consumer polls records and then the resulting data gets processed.

If you do wish to create transactions, you need to enable a `kafka-streams-spans` module:

```yml
class_transformer:
  kafka-streams-spans:
    enabled: true
```

## Enable Kafka Connect transaction [#collect-kafka-connect-transactions]

If you're using Kafka Connect, by default we do not enable transactions. This is to prevent unnecessary overhead because Kafka applications tend to have high throughput.

Kafka Connect transactions are recorded for each iteration of the sink/source task. For a sink task, a transaction consists of polling a Kafka consumer, converting each message, and sending data to the target. For a source task, a transaction consists of reading from the target, converting the data into messages, and sending each message with a Kafka producer.

If you do wish to collect these transactions, you need to enable a `kafka-connect-spans` module:

```yml
class_transformer:
  kafka-connect-spans:
    enabled: true
```

## Kafka distributed traces [#collect-kafka-distributed-traces]

The Java agent can also collect [distributed traces](https://docs.newrelic.com/docs/apm/distributed-tracing/getting-started/introduction-distributed-tracing) from Kafka clients. Because Kafka Streams runs on top of Kafka clients, the steps to manage distributed tracing also apply. Enabling traces doesn't affect the agent's normal operations; it will still report metric or event data from Kafka.

Impacts and requirements to consider before enabling:

-   **The instrumentation adds a 150 to 200 byte payload to message headers.** If your Kafka messages are very small, traces can add significant processing and storage overhead. This additional payload size could cause Kafka to drop messages if they exceed your Kafka messaging size limit. For this reason, we recommend testing out Kafka distributed traces in a dev environment before enabling them in production.
-   Distributed tracing is only available for Kafka client versions 0.11.0.0 or higher.
-   If you have **not** enabled distributed tracing for your app before, read the [Transition guide](https://docs.newrelic.com/docs/apm/distributed-tracing/getting-started/transition-guide-distributed-tracing) before enabling.
-   To propagate W3C trace context via Kafka message headers, see the [distributed tracing API usage guide](https://docs.newrelic.com/docs/agents/java-agent/api-guides/guide-using-java-agent-api#trace-calls) for details on APIs that were released in Java agent 6.4.0. Note that adding additional headers to Kafka messages will further increase the payload size. To see these APIs in action, see [Using Java agent trace APIs with Kafka](https://github.com/newrelic/newrelic-java-examples/tree/main/newrelic-java-agent/distributed-tracing/kafka-examples).
-   If you're using Kafka Streams, you need to enable a span instrumentation module (refer to the [Kafka Streams transaction section](#collect-kafka-streams-transactions)). Because a transaction is not recorded per record, accepting distributed trace headers will only work for one record.

The complete process of enabling this is described below but at a high level it involves these basic steps: 1) enable tracing via the agent config, and 2) call the [Java agent API](https://docs.newrelic.com/docs/agents/java-agent/api-guides/guide-using-java-agent-api) to instrument transactions on both the producer and consumer side.

To collect distributed traces from Kafka:

**1. Enable distributed tracing in the config file**

If you have not enabled distributed tracing for your app before, read the [distributed tracing transition guide](https://docs.newrelic.com/docs/apm/distributed-tracing/getting-started/transition-guide-distributed-tracing) before you enable it.

To enable Kafka distributed tracings, these two settings must be enabled in your [`newrelic.yml` config file](https://docs.newrelic.com/docs/agents/java-agent/configuration/java-agent-configuration-config-file#Structure):

-   Make sure the [`distributed_tracing`](https://docs.newrelic.com/docs/agents/java-agent/configuration/java-agent-configuration-config-file#distributed-tracing) element is enabled:

    ```yml
    distributed_tracing:
      enabled: true
    ```
-   Enable general Kafka-specific distributed tracing features by adding the following to your config file:

    ```yml
    class_transformer:
      com.newrelic.instrumentation.kafka-clients-spans-0.11.0.0:
        enabled: true
    ```
-   Enable spring-kafka instrumentation for agent versions `8.21` and up.

    > #### ⚠️ IMPORTANT
    >
    > This may cause a spike in ingest and memory usage so it is disabled

    ```yml
    class_transformer:
      com.newrelic.instrumentation.spring-kafka-2.2.0:
        enabled: true
    ```
-   Enable distributed tracing for batch calls for Kafka for agent versions `8.21` and up. (Batch call distributed traces are disabled by default as transactions can only accept 1 parent trace from a message):
    ```yml
    kafka:
      spans:
        distributed_trace: 
          consume_many:
            enabled: true
    ```

**2. Instrument the Kafka producer**

To instrument your Kafka producer, you'll need to start a transaction before any calls to `Producer.send(ProducerRecord<K, V> record)`. To do this, add the Java agent `@Trace(dispatcher = true)` annotation to the method.

For example:

````java
@Trace(dispatcher = true)
public static void createAndSend(KafkaProducer<String, String> producer){
  ProducerRecord<String, String> data = new ProducerRecord<String, String>("topic", "key", "value");
  producer.send(data);
}
```

<Callout variant="important">
  If you're using Kafka Streams, you don't need to directly start a transaction or send data to a producer.
</Callout>

````

**3. Instrument the Kafka consumer**

### Scenario 1: Using manual instrumentation with Kafka Client

To instrument your Kafka consumer via manual insteumantation, you'll need to start a transaction when the message is being processed. The agent stores the distributed tracing payload header under the `newrelic` key or under the W3C's `traceparent` and `tracestate` keys. Retrieve the header, then call the New Relic transaction API to accept the payload.

For example:

````java
@Trace(dispatcher = true)
private static void processMessage(ConsumerRecord<String, String> rec) {
  // create a distributed trace headers map
  Headers dtHeaders = ConcurrentHashMapHeaders.build(HeaderType.MESSAGE);

  // Iterate through each record header and insert the trace headers into the dtHeaders map
  for (Header header : rec.headers()) {
    String headerValue = new String(header.value(), StandardCharsets.UTF_8);

    // using the newrelic key
    if (header.key().equals("newrelic")) {
      dtHeaders.addHeader("newrelic", headerValue);
    }

    // or using the W3C keys
    if (header.key().equals("traceparent")) {
      dtHeaders.addHeader("traceparent", headerValue);
    }

    if (header.key().equals("tracestate")) {
      dtHeaders.addHeader("tracestate", headerValue);
    }
  }

  // Accept distributed tracing headers to link this request to the originating request
  NewRelic.getAgent().getTransaction().acceptDistributedTraceHeaders(TransportType.Kafka, dtHeaders);
}
```
### Scenario 2: Using auto instrumentation for kafka clients
This will only apply for agent versions `8.21` and up. You can leverage auto instrumentation too to read distributed tracing for kafka consumers. 
This is accomplished by starting a transaction and calling the `poll(...)` method of a consumer when the transaction is happening.

This is disabled by dafault because consumer polling by default accepts multiple messages and transactions can only have one parent trace so only 1 message will be read.

To enable automatically accepting distributed traces for Kafka consumer polls, you need to enable distributed tracing for batch calls as showin in step 1.

Example of instrumenting kafka consumer polls:

```java
@Trace(dispatcher = true)
private void consumeMessage(KafkaConsumer<String, String> consumer) {
  final ConsumerRecords<String, String> records = consumer.poll(1000);
  // Process your records ....
}
```

### Scenario 3: Using Spring Kafka
If you are using Spring Kafka, you can enable auto instrumentation when for methods instrumented with `@KafkaListener` if you are using agent versions `8.21.0` and up.
This creates transactions and auto-acceots distributed traces for consumers.
Due to Kafka being a high throughput platform, it is disabled to prevent too much ingest and memory ovehead. 
You can enable spring-kafka instrunenation with the config shown in step 1.

Example:
```java
@KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId")
public void listen(String data) {
  ...
}
```

<Callout variant="important">
  If you enabled transactions with Kafka Streams (refer to the [Kafka Streams transaction section](#collect-kafka-streams-transactions)),
  even though a transaction applies to many records, accepting distributed trace headers will only apply to one record.
</Callout>

````