--- title: Kafka Metrics description: Datadog, the leading service for cloud-scale monitoring. breadcrumbs: Docs > OpenTelemetry in Datadog > Integrations > Kafka Metrics --- > For the complete documentation index, see [llms.txt](https://docs.datadoghq.com/llms.txt). # Kafka Metrics ## Overview{% #overview %} {% image source="https://docs.dd-static.net/images/opentelemetry/collector_exporter/kafka_metrics.34bf4de835ecb56a140a92077ecf86cd.png?auto=format&fit=max&w=850 1x, https://docs.dd-static.net/images/opentelemetry/collector_exporter/kafka_metrics.34bf4de835ecb56a140a92077ecf86cd.png?auto=format&fit=max&w=850&dpr=2 2x" alt="OpenTelemetry Kafka metrics in the Kafka dashboard" /%} The [Kafka metrics receiver](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/receiver/kafkametricsreceiver) and the [JMX Scraper](https://github.com/open-telemetry/opentelemetry-java-contrib/tree/main/jmx-scraper) collect Kafka metrics for the out-of-the-box [Kafka Dashboard](https://app.datadoghq.com/dash/integration/50/kafka-zookeeper-and-kafka-consumer-overview), **Kafka, Zookeeper, and Kafka Consumer Overview**. The two collect different data and most setups run both. The Kafka metrics receiver reads cluster state through the Kafka admin API. The JMX Scraper reads JMX beans exposed by brokers, producers, and consumers. ## Kafka metrics receiver{% #kafka-metrics-receiver %} Set `KAFKA_BROKER_ADDRESS` to a broker address in `host:port` format. To list more than one broker, separate the addresses with commas. {% tab title="Host" %} Add the receiver to the Collector configuration: ```yaml receivers: kafkametrics: brokers: "${env:KAFKA_BROKER_ADDRESS}" protocol_version: 2.0.0 scrapers: - brokers - topics - consumers ``` Add `kafkametrics` to the `receivers` list of the metrics pipeline. The Collector starts only the receivers that a pipeline uses. For example, in the recommended Host configuration: ```yaml service: pipelines: metrics: receivers: [otlp, host_metrics, kafkametrics] processors: [resource_detection, cumulativetodelta] exporters: [otlp_http] ``` {% /tab %} {% tab title="Kubernetes" %} Run the Kafka metrics receiver in a single-replica Collector deployment to prevent duplicate metrics. The Collector can export metrics directly to Datadog over OTLP HTTP or forward them to another Collector. Set the deployment mode in `values.yaml`: ```yaml mode: deployment ``` Add the receiver to the Collector configuration: ```yaml receivers: kafkametrics: brokers: ${env:KAFKA_BROKER_ADDRESS} protocol_version: 2.0.0 scrapers: - brokers - topics - consumers ``` Add `kafkametrics` to a metrics pipeline in the same configuration. The Collector starts only the receivers that a pipeline uses. To export directly to Datadog, configure the `otlp_http` exporter as shown in the [recommended setup](https://docs.datadoghq.com/opentelemetry/setup/collector_exporter.md#otlp-http-exporter): ```yaml service: pipelines: metrics: receivers: [kafkametrics] exporters: [otlp_http] ``` {% /tab %} ## JMX Scraper{% #jmx-scraper %} The JMX Scraper runs as a standalone Java application and sends metrics to the Collector's OTLP receiver. The [recommended Collector setup](https://docs.datadoghq.com/opentelemetry/setup/collector_exporter.md) receives OTLP over gRPC on port 4317 and includes the OTLP receiver in its metrics pipeline, so you don't need to change the receiver configuration. The JMX Scraper sends counters with cumulative temporality. Keep the `cumulativetodelta` processor in the recommended metrics pipeline, which converts them to the delta temporality that the [Datadog OTLP metrics intake](https://docs.datadoghq.com/opentelemetry/setup/otlp_ingest/metrics.md#ensure-only-delta-metrics-are-sent) requires. ### Configure Kafka brokers{% #configure-kafka-brokers %} 1. Add the following JVM options to each Kafka broker to start an unsecured JMX endpoint suitable for evaluation: ```shell -Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.port=9999 -Dcom.sun.management.jmxremote.rmi.port=9999 -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname= ``` Important alert (level: warning): These options disable JMX authentication and TLS. Use this configuration only for evaluation. For production, enable authentication and TLS and configure the scraper with credentials and trust store settings. `jmxremote.authenticate` and `jmxremote.ssl` default to `true`, so set both properties explicitly for this evaluation configuration. Without a password file, authentication causes the broker to fail with `Password file not found`. When `jmxremote.rmi.port` matches `jmxremote.port`, SSL causes `Port already in use` because the RMI registry is not protected with TLS and cannot share the port. To enable authentication and TLS for production, add `com.sun.management.jmxremote.registry.ssl=true` so the registry and connector can share a port. Configure the scraper with the `otel.jmx.username`, `otel.jmx.password`, `otel.jmx.remote.registry.ssl`, `javax.net.ssl.trustStore`, and `javax.net.ssl.trustStorePassword` properties. 1. Set `java.rmi.server.hostname` to the hostname or IP address that the scraper uses to connect. The broker includes this value in an RMI stub, and the scraper opens a second connection to it. - Use a loopback address only when the scraper runs on the broker's own host or pod. Several Kafka distributions and container images, including Confluent Platform, default to loopback. - Do not use `0.0.0.0`. It is a bind address and the scraper cannot connect to it. - In Kubernetes, use the broker pod IP or a DNS name. In Kafka container images, `KAFKA_JMX_HOSTNAME` sets `java.rmi.server.hostname`. Set it to the same address. An unreachable address can allow the initial connection to succeed, but the scrape then fails with `Failed to retrieve RMIServer stub` or `Connection refused to host: `. 1. Set `jmxremote.rmi.port` to the same port as `jmxremote.port`. Without a fixed RMI port, the second connection uses a random port that a firewall or port mapping might block. To scrape producers and consumers, add the same JVM options to those applications. ### Run the JMX Scraper{% #run-the-jmx-scraper %} Set `otel.jmx.target.source` to `legacy` to preserve the JVM metric names that existing dashboards and monitors expect. Without it, the scraper defaults to `auto` and every JVM metric name changes. {% tab title="Host" %} Install a JRE on the host that runs the JMX Scraper. On Debian or Ubuntu, run: ```shell apt-get update && \ apt-get -y install default-jre-headless ``` Download the JMX Scraper JAR from the [OpenTelemetry Java Contrib releases](https://github.com/open-telemetry/opentelemetry-java-contrib/releases). Set each address variable in `host:port` format. Each scraper process scrapes a single JMX endpoint. Run one process for each broker, producer, and consumer, with that JVM's address: ```shell # Kafka broker java \ -Dotel.jmx.service.url=service:jmx:rmi:///jndi/rmi://${KAFKA_BROKER_JMX_ADDRESS}/jmxrmi \ -Dotel.jmx.target.system=kafka,jvm \ -Dotel.jmx.target.source=legacy \ -Dotel.metrics.exporter=otlp \ -Dotel.exporter.otlp.endpoint=http://localhost:4317 \ -jar /path/to/opentelemetry-jmx-scraper.jar # Kafka producer java \ -Dotel.jmx.service.url=service:jmx:rmi:///jndi/rmi://${KAFKA_PRODUCER_JMX_ADDRESS}/jmxrmi \ -Dotel.jmx.target.system=kafka-producer \ -Dotel.jmx.target.source=legacy \ -Dotel.metrics.exporter=otlp \ -Dotel.exporter.otlp.endpoint=http://localhost:4317 \ -jar /path/to/opentelemetry-jmx-scraper.jar # Kafka consumer java \ -Dotel.jmx.service.url=service:jmx:rmi:///jndi/rmi://${KAFKA_CONSUMER_JMX_ADDRESS}/jmxrmi \ -Dotel.jmx.target.system=kafka-consumer \ -Dotel.jmx.target.source=legacy \ -Dotel.metrics.exporter=otlp \ -Dotel.exporter.otlp.endpoint=http://localhost:4317 \ -jar /path/to/opentelemetry-jmx-scraper.jar ``` {% /tab %} {% tab title="Kubernetes" %} Use the following Dockerfile to build an image with a JRE and the JMX Scraper JAR: ```Dockerfile FROM alpine:latest AS prep # JMX Scraper JAR ARG JMX_SCRAPER_JAR_VERSION=1.60.0 ADD https://github.com/open-telemetry/opentelemetry-java-contrib/releases/download/v${JMX_SCRAPER_JAR_VERSION}/opentelemetry-jmx-scraper.jar /opt/opentelemetry-jmx-scraper.jar ARG USER_UID=65532 RUN chown ${USER_UID} /opt/opentelemetry-jmx-scraper.jar FROM gcr.io/distroless/java17-debian11:nonroot COPY --from=prep /opt/opentelemetry-jmx-scraper.jar /opt/opentelemetry-jmx-scraper.jar ENTRYPOINT ["java"] CMD ["-Dotel.jmx.service.url=service:jmx:rmi:///jndi/rmi://kafka:9999/jmxrmi", \ "-Dotel.jmx.target.system=kafka,jvm", \ "-Dotel.jmx.target.source=legacy", \ "-Dotel.metrics.exporter=otlp", \ "-Dotel.exporter.otlp.endpoint=http://otelcol:4317", \ "-jar", \ "/opt/opentelemetry-jmx-scraper.jar"] ``` Replace `kafka:9999` with the broker's JMX address and `otelcol:4317` with the Collector's OTLP gRPC endpoint. To scrape producers or consumers, set `otel.jmx.target.system` to `kafka-producer` or `kafka-consumer` and point `otel.jmx.service.url` at that JVM's JMX endpoint. Run a single instance per JMX endpoint. Multiple replicas scraping the same endpoint produce duplicate metrics. {% /tab %} ## Log collection{% #log-collection %} To collect logs with the OpenTelemetry Collector, see [Log Collection](https://docs.datadoghq.com/opentelemetry/config/log_collection.md). To include Kafka logs in the out-of-the-box Kafka Dashboard, use an attributes processor to add the `source:kafka` tag: ```yaml processors: attributes: actions: - key: ddtags value: "source:kafka" action: insert ``` To add this attribute only to Kafka logs, use [include/exclude filtering](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/attributesprocessor/README.md#includeexclude-filtering) in the attributes processor. ## Data collected{% #data-collected %} ### Kafka metrics receiver{% #kafka-metrics-receiver-1 %} | OTEL | DESCRIPTION | | ------------------------------ | --------------------------------------------------------------- | | kafka.consumer_group.lag | Current approximate lag of consumer group at partition of topic | | kafka.consumer_group.offset | Current offset of the consumer group at partition of topic | | kafka.partition.current_offset | Current offset of partition of topic. | ### JMX Scraper{% #jmx-scraper-1 %} #### Kafka broker{% #kafka-broker %} | OTEL | DESCRIPTION | FILTER | | -------------------------------- | ------------------------------------------------------------------- | ----------------------- | | kafka.controller.active.count | Controller is active on broker | | kafka.isr.operation.count | The number of in-sync replica shrink and expand operations | `operation`: `shrink` | | kafka.isr.operation.count | The number of in-sync replica shrink and expand operations | `operation`: `expand` | | kafka.leader.election.rate | Leader election rate - increasing indicates broker failures | | kafka.logs.flush.time.count | Log flush count | | kafka.max.lag | Max lag in messages between follower and leader replicas | | kafka.message.count | The number of messages received by the broker | | kafka.network.io | The bytes received or sent by the broker | `state`: `in` | | kafka.network.io | The bytes received or sent by the broker | `state`: `out` | | kafka.partition.count | The number of partitions on the broker | | kafka.partition.offline | The number of partitions offline | | kafka.partition.under_replicated | The number of under replicated partitions | | kafka.purgatory.size | The number of requests waiting in purgatory | `type`: `fetch` | | kafka.purgatory.size | The number of requests waiting in purgatory | `type`: `produce` | | kafka.request.failed | The number of requests to the broker resulting in a failure | `type`: `produce` | | kafka.request.failed | The number of requests to the broker resulting in a failure | `type`: `fetch` | | kafka.request.queue | Size of the request queue | | kafka.request.time.99p | The 99th percentile time the broker has taken to service requests | `type`: `produce` | | kafka.request.time.99p | The 99th percentile time the broker has taken to service requests | `type`: `fetchfollower` | | kafka.request.time.99p | The 99th percentile time the broker has taken to service requests | `type`: `fetchconsumer` | | kafka.request.time.avg | The average time the broker has taken to service requests | `type`: `produce` | | kafka.request.time.avg | The average time the broker has taken to service requests | `type`: `fetchconsumer` | | kafka.request.time.avg | The average time the broker has taken to service requests | `type`: `fetchfollower` | | kafka.unclean.election.rate | Unclean leader election rate - increasing indicates broker failures | #### Kafka producer{% #kafka-producer %} | OTEL | DESCRIPTION | | ---------------------------------- | ----------------------------------------------------------------------------------------------- | | kafka.producer.byte-rate | The average number of bytes sent per second for a topic. | | kafka.producer.compression-rate | The average compression rate of record batches for a topic. | | kafka.producer.io-wait-time-ns-avg | The average length of time the I/O thread spent waiting for a socket ready for reads or writes. | | kafka.producer.outgoing-byte-rate | The average number of outgoing bytes sent per second to all servers. | | kafka.producer.record-error-rate | The average per-second number of record sends that resulted in errors for a topic. | | kafka.producer.record-retry-rate | The average per-second number of retried record sends for a topic. | | kafka.producer.record-send-rate | The average number of records sent per second for a topic. | | kafka.producer.request-latency-avg | The average request latency. | | kafka.producer.request-rate | The average number of requests sent per second. | | kafka.producer.response-rate | Responses received per second. | #### Kafka consumer{% #kafka-consumer %} | OTEL | DESCRIPTION | | ------------------------------------------ | ----------------------------------------------------------------- | | kafka.consumer.bytes-consumed-rate | The average number of bytes consumed per second | | kafka.consumer.fetch-size-avg | The average number of bytes fetched per request | | kafka.consumer.records-consumed-rate | The average number of records consumed per second | | kafka.consumer.total.bytes-consumed-rate | The average number of bytes consumed for all topics per second. | | kafka.consumer.total.records-consumed-rate | The average number of records consumed for all topics per second. | **Note**: Datadog replaces hyphens (`-`) with underscores (`_`) in metric names. For example, `kafka.producer.request-rate` becomes `kafka.producer.request_rate`. For the complete mapping between OpenTelemetry and Datadog metric names, see [OpenTelemetry Metrics Mapping](https://docs.datadoghq.com/opentelemetry/mapping/metrics_mapping.md). ## Example logging output{% #example-logging-output %} ``` Metric #0 Descriptor: -> Name: kafka.message.count -> Description: The number of messages received by the broker -> Unit: {messages} -> DataType: Sum -> IsMonotonic: true -> AggregationTemporality: Cumulative NumberDataPoints #0 StartTimestamp: 2024-01-22 15:50:24.212 +0000 UTC Timestamp: 2024-01-22 15:51:24.218 +0000 UTC Value: 25 ```