Java agent: Instrument Kafka message queues

The New Relic Java agent automatically collects data from Kafka'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 three types of Kafka data:

• Kafka metrics
• Kafka events
• Enable Kafka Streams transactions
• Kafka distributed traces

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. 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, the agent automatically reports rich Kafka metrics with information about messaging rates, latency, lag, and more. The Java agent collects all Kafka consumer and producer metrics (but not connect or stream metrics).

To view these metrics, create a custom dashboard:

1. Go to the New Relic metric explorer.

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.

Enable Kafka event collection

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). This allows you to use 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 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 captured since the previous event. The event is also collected every 30 seconds.

To enable Kafka event collection:

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

   kafka.metrics.as_events.enabled: true

   카피

2. Restart your JVM.

3. Use the event explorer to view your Kafka events, located in the KafkaMetrics event type. Or, use NRQL to query your events directly. For example:

   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.

Enable Kafka node metrics

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:

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

Enable Kafka config events

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:

class_transformer:
  kafka-clients-config:
    enabled: true

Enable 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:

class_transformer:
  kafka-streams-spans:
    enabled: true

Enable Kafka Connect transaction

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:

class_transformer:
  kafka-connect-spans:
    enabled: true

Enable Kafka distributed traces

The Java agent can also collect distributed traces 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 before enabling.
• To propagate W3C trace context via Kafka message headers, see the distributed tracing API usage guide 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.
• If you're using Kafka Streams, you need to enable a span instrumentation module (refer to the Kafka Streams transaction section). 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 to instrument transactions on both the producer and consumer side.

To collect distributed traces from Kafka:

@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);
}

@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);
}