Skip to main content
The Apache Kafka Consumer integration requires CXDOT Collector 1.10.0 or greater. Apache Kafka is an open source distributed event streaming platform. Use the Kafka Consumer integration to collect consumer group offsets and lag, partition offsets and replica counts, and broker counts from the Kafka clusters running in your environment. The Kafka Consumer integration supports Apache Kafka 3.8 or greater.

​
Supported telemetry types

The Kafka Consumer integration supports these telemetry types:

​
Prerequisites

The Kafka Consumer integration has the following prerequisites:
  • Make every broker in the cluster reachable from the CXDOT Collector at the address the broker advertises. The Collector connects to the cluster through one bootstrap address, learns the other brokers from the cluster metadata, and then connects to each broker at its advertised listener. If the Collector can’t reach one of the brokers, it reports no consumer group metrics for the cluster in that collection interval. To list the advertised addresses, run the kafka-broker-api-versions.sh tool from the bin directory of a Kafka installation, on a host or pod with the same network access as the CXDOT Collector:
    Replace BOOTSTRAP_ADDRESS with the host:port that the integration connects to. The output lists each broker by its advertised address and ID. Confirm that no broker’s entry reports ERROR. For a cluster that requires authentication or TLS, pass the client settings in a properties file with the --command-config option.
  • If your cluster requires authentication, prepare credentials for SASL PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512, for Amazon Managed Streaming for Apache Kafka (MSK) Identity and Access Management (IAM) authentication, or for Kerberos. If your cluster enforces authorization, grant the principal that the Collector authenticates as the Describe operation on the topics and consumer groups to report. For more information, see Authorization in the Kafka documentation. With MSK IAM authentication, Kafka access control lists don’t apply to IAM identities, so grant the equivalent permissions in an IAM policy instead.

​
Configure

The Kafka Consumer integration is enabled by default and monitors each Kafka cluster that you identify with an autodiscovery annotation or a static target. Each target reports on a whole cluster through any one of its brokers, so configure exactly one target for each cluster. With a target for each broker, the integration reports every time series once per broker. To configure the Kafka Consumer integration, follow these steps:
  1. For a Kafka cluster running in your Kubernetes cluster, add autodiscovery annotations for Kafka Consumer to the Service in front of the brokers, such as a bootstrap Service, and supply the bootstrap address and any credentials in the annotation. One elected replica of the cluster-tier CXDOT Collector monitors each annotated Service. The annotation also selects the consumer groups and topics to report, and carries the connection’s TLS and authentication settings. For more information, see autodiscovery and architecture. Annotate the Service, not the broker pods. Each annotated broker pod becomes its own target for the whole cluster. For MSK IAM authentication, set the AWS Region in the annotation’s aws_region setting. The Collector doesn’t detect the Region, so without it the connection fails authentication.
  2. Optional: Configure static targets instead of discovered targets, such as Kafka clusters outside your Kubernetes cluster. Add one entry for each cluster, and point it at a bootstrap address that resolves to several brokers, such as a DNS name, a Kubernetes Service, or a load balancer. For example, add the following to the values.yaml file for the CXDOT Collector Helm chart:
    This integration instance monitors exactly the listed clusters and doesn’t discover annotated targets. In a Kubernetes deployment, one elected replica of the cluster-tier CXDOT Collector monitors static targets.
  3. Optional: Configure separate integration instances to collect from both discovered and static targets. For example, add the following to the values.yaml file for the CXDOT Collector Helm chart:
    The kafka_consumer/external key defines a named instance for the static targets. Configuring a named instance suppresses the implicit default instance, so the empty kafka_consumer: {} entry is required to keep discovering annotated targets.
  4. Optional: If your static targets require authentication or TLS, configure the auth and tls settings, and provide the password through your deployment’s secret management. For example, the following Helm values authenticate with SCRAM-SHA-512, verify the brokers’ certificates against a certificate authority file mounted into the Collector, and reference the KAFKA_CONSUMER_SASL_PASSWORD environment variable:
    The auth setting also accepts PLAIN, SCRAM-SHA-256, MSK IAM authentication, and Kerberos. The Collector masks passwords in its debug output and logs. For annotated targets, supply credentials and TLS settings in the annotation instead.
  5. Optional: Limit what the integration collects from clusters with many topics, partitions, or consumer groups. The integration reports several time series for each partition of every topic that topic_match admits, and requests the offsets of all of those partitions on every collection. For static targets, narrow topic_match and group_match. For example, the following Helm values report only topics whose names start with orders. and consumer groups whose names start with orders-:
    Annotated targets take their topic and consumer group selection from the annotation. To limit collection for any target, including annotated ones, remove topics from scrapers to stop collecting topic and partition metrics, or turn off individual metrics with metrics.
  6. Optional: Disable the integration. For example, add the following to the values.yaml file for the CXDOT Collector Helm chart:

​
Validate

To validate the Kafka Consumer integration, follow these steps:
  1. In the Live Telemetry Analyzer, add the following filters:
    • __name__=cxdot.integration.target.health
    • cxdot.integration.name=kafka_consumer
    Confirm that each monitored cluster reports 1. A value of 0 means the CXDOT Collector hit an error collecting from that cluster. The cxdot.integration.target attribute identifies the cluster by the bootstrap address the Collector connects to.
  2. In Metrics Explorer, run the following query:
    Confirm that the query returns a time series for each consumer group and each topic where the group has committed offsets, with the group’s total lag across the topic’s partitions as the value. The kafka.cluster.id attribute identifies the cluster.

​
Troubleshooting

  • The health metric reports 1, but consumer group metrics are missing or intermittent: The health metric needs a response from only one broker, but consumer group metrics need a connection to every broker. When the Collector can’t reach one broker at its advertised address, it drops all consumer group metrics for that collection interval, along with the offsets of the partitions that broker leads. Check the advertised addresses as described in the Prerequisites section.
  • Consumer group and partition time series appear once for each broker pod: More than one target monitors the same cluster, typically because the annotations are on the broker pods instead of the cluster’s Service. Annotate the Service instead, and list each cluster once among static targets.
  • A consumer group reports members but no offsets or lag, or lag is missing for some partitions: The integration reports offsets and lag only for partitions where the group has committed an offset. A partition the group hasn’t committed to has no lag time series, rather than a lag of 0. Lag is the partition’s high watermark minus the group’s committed offset, and is never less than 0.

​
Configuration reference

Configure one Apache Kafka Consumer integration instance with the following settings. In Helm values, place these settings under config.integrations.kafka_consumer. In a Collector configuration file, place them under cxdot.integrations.kafka_consumer.

​
Optional settings

  • enabled Type: boolean. Optional. Default: true. Whether to enable this Kafka Consumer integration instance. If true, the Collector runs the instance. If false, the Collector doesn’t run it.
  • endpoints Type: array of object. Optional. Default: []. Static Kafka clusters to monitor, one entry per cluster. Each endpoint is the host:port of one broker in that cluster. The Collector learns the remaining brokers from cluster metadata and must be able to reach every broker at the address it advertises. If it can’t reach one of them, it reports no consumer group metrics for that cluster in that collection interval. The Collector first reaches the cluster through this one address, so point each endpoint at a bootstrap address that resolves to several brokers, such as a DNS name, a Kubernetes Service, or a load balancer. An outage of a single broker when the Collector starts then doesn’t leave the cluster unmonitored. Listing several brokers of the same cluster reports that cluster once per entry. When this list is empty or omitted, the integration uses automatic target discovery. When the list contains clusters, this integration instance monitors exactly those clusters and doesn’t automatically discover targets. Configure a second named instance to use both methods.
  • endpoints[].endpoint Type: string. Required. host:port of the cluster’s bootstrap address.
  • collection_interval Type: duration. Optional. Default: 15s. How often the Collector collects metrics from each Kafka cluster. Targets discovered through annotations don’t use this setting or timeout. Their annotations determine how often the Collector collects from them.
  • timeout Type: duration. Optional. Default: 15s. How long one collection from a cluster can run before the Collector abandons it. One collection makes several requests to the brokers, and this setting bounds all of them together. Keep it no longer than collection_interval so that a slow collection doesn’t hold up the next one.
  • initial_delay Type: duration. Optional. Default: 1s. How long the Collector waits after starting before it first collects metrics.
  • scrapers Type: array of string. Optional. Default: ["brokers","topics","consumers"]. Minimum items: 1. Which groups of metrics to collect. brokers reports the number of brokers in the cluster. topics reports partition counts, partition offsets, and replica counts. It produces several time series for each partition of every topic that topic_match admits, and the Collector requests the offsets of all of those partitions on every collection. consumers reports the members of each consumer group, and the committed offset and lag of each partition where the group has committed an offset. Lag is the partition’s high watermark, which kafka.partition.current_offset also reports, minus the group’s committed offset, and is never less than 0.
  • topic_match Type: regex. Optional. Default: ^[^_].*$. Regular expression selecting the topics to report, both for partition metrics and for consumer group offsets and lag. The expression uses RE2 syntax and matches any part of a topic name unless you anchor it with ^ and $. The default excludes topics whose names start with an underscore, which Kafka and common platform components use for internal topics. With topics in scrapers, the Collector reports several time series for each partition of every matched topic and requests all of their offsets on every collection. A narrower expression lowers that cost on clusters with many partitions. Targets discovered through annotations don’t use this setting. Their annotations determine which topics they report. For those targets, scrapers and metrics are the settings that still limit what the Collector collects.
  • group_match Type: regex. Optional. Default: .*. Regular expression selecting the consumer groups to report. The expression uses RE2 syntax and matches any part of a group name unless you anchor it with ^ and $. The default reports every consumer group in the cluster. This setting doesn’t narrow the partition metrics: with topics in scrapers, the Collector still reports several time series for each partition of every topic that topic_match admits and requests all of their offsets on every collection. Targets discovered through annotations don’t use this setting. Their annotations determine which consumer groups they report.
  • cluster_alias Type: string. Optional. Default: “. A name for the cluster, reported as the kafka.cluster.alias resource attribute when resource_attributes.cluster_alias.enabled is true.
  • client_id Type: string. Optional. Default: otel-collector. Minimum length: 1. Client ID the Collector presents to the brokers, which they can use for access control lists, quotas, and request logging.
  • protocol_version Type: string. Optional. Default: “. Kafka version whose protocol the Collector speaks, such as 3.8.0. When empty, the Collector negotiates the protocol version with each broker.
  • conn_idle_timeout Type: duration. Optional. Default: 9m. Roughly how long the Collector keeps an idle connection to a broker open before closing it.
  • auth Type: object. Optional. Authentication for statically configured clusters, with either SASL (sasl) or Kerberos (kerberos). When omitted, the Collector connects without authenticating. The Collector rejects the deprecated plain_text and auth.tls settings: use sasl with the PLAIN mechanism and the top-level tls setting instead. Targets discovered through annotations use the authentication settings in their annotations.
  • auth.sasl Type: object. Optional. SASL authentication settings.
  • auth.sasl.mechanism Type: string. Optional. SASL mechanism the Collector authenticates with. Set it whenever you configure sasl. PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512 authenticate with username and password. AWS_MSK_IAM_OAUTHBEARER authenticates to Amazon Managed Streaming for Apache Kafka (MSK) with Identity and Access Management (IAM) credentials from Amazon Web Services (AWS). The Collector loads those credentials from the default AWS credential chain, such as environment variables or a role for its Kubernetes service account, and signs them for the region in aws_msk.region. MSK clients that use IAM authentication connect over TLS, so configure tls as well. Allowed values: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, AWS_MSK_IAM_OAUTHBEARER.
  • auth.sasl.password Type: string. Optional. Password for the PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512 mechanisms, which require it. The Collector masks this value in its debug output and logs.
  • auth.sasl.aws_msk Type: object. Optional. Settings for the AWS_MSK_IAM_OAUTHBEARER mechanism.
  • auth.sasl.aws_msk.region Type: string. Optional. AWS Region of the MSK cluster, such as us-east-1.
  • auth.sasl.username Type: string. Optional. Username for the PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512 mechanisms, which require it.
  • auth.kerberos Type: object. Optional. Kerberos authentication settings. The Collector authenticates with the keytab in keytab_file when that setting isn’t empty, and with password otherwise.
  • auth.kerberos.password Type: string. Optional. Password for username, used when keytab_file is empty. The Collector masks this value in its debug output and logs.
  • auth.kerberos.config_file Type: string. Optional. Path to the Kerberos configuration file, usually krb5.conf, that lists the key distribution centers for realm.
  • auth.kerberos.disable_fast_negotiation Type: boolean. Optional. Whether to leave Flexible Authentication Secure Tunneling (FAST) negotiation out of the ticket requests that the Collector sends. If true, the Collector leaves it out. Set it to true when your key distribution center rejects that negotiation. If false, the Collector includes it.
  • auth.kerberos.keytab_file Type: string. Optional. Path to a keytab file that holds the key for username. When set, the Collector authenticates with the keytab instead of password.
  • auth.kerberos.realm Type: string. Optional. Realm of username, such as EXAMPLE.COM.
  • auth.kerberos.service_name Type: string. Optional. Service name in the brokers’ principals, such as kafka.
  • auth.kerberos.use_keytab Type: boolean. Optional. The Collector accepts this setting but ignores it. Whether the Collector authenticates with a keytab depends only on keytab_file.
  • auth.kerberos.username Type: string. Optional. Principal name the Collector authenticates as, without the realm.
  • tls Type: object. Optional. Transport Layer Security (TLS) settings for statically configured clusters. When this block is present, even empty, the Collector connects to the brokers over TLS and verifies their certificates against the system certificate authority pool, unless other tls settings change that. When omitted, the Collector connects without TLS. Targets discovered through annotations use the TLS settings in their annotations.
  • tls.key_pem Type: string. Optional. PEM-encoded client private key content used with cert_pem for mutual TLS authentication. The Collector masks this value in its debug output and logs.
  • tls.tpm Type: object. Optional. Trusted platform module settings for the TLS private key.
  • tls.tpm.auth Type: string. Optional. Authorization value for the trusted platform module key.
  • tls.tpm.owner_auth Type: string. Optional. Owner authorization value for the trusted platform module.
  • tls.tpm.enabled Type: boolean. Optional. Whether to use a trusted platform module for the TLS private key. If true, the Collector loads the key from the configured device or socket. If false, the Collector uses the configured key file or in-memory key.
  • tls.tpm.path Type: string. Optional. Path to the trusted platform module device or Unix domain socket, such as /dev/tpm0 or /dev/tpmrm0.
  • tls.ca_file Type: string. Optional. Path to a PEM-encoded certificate authority (CA) certificate file used to verify the brokers’ certificates. If omitted, the Collector uses the system certificate authority pool.
  • tls.ca_pem Type: string. Optional. PEM-encoded certificate authority certificate content used to verify the brokers’ certificates.
  • tls.cert_file Type: string. Optional. Path to the PEM-encoded client certificate file to present when the brokers require mutual TLS authentication.
  • tls.cert_pem Type: string. Optional. PEM-encoded client certificate content to present when the brokers require mutual TLS authentication.
  • tls.cipher_suites Type: array of string. Optional. TLS cipher suites that the Collector can use, in preference order. If omitted, the Collector uses the secure defaults from the Go TLS library. For supported names, see https://go.dev/src/crypto/tls/cipher_suites.go.
  • tls.curve_preferences Type: array of string. Optional. Elliptic curves that the Collector can use for an elliptic curve Diffie-Hellman ephemeral handshake, in preference order. If omitted, the Collector uses the Go TLS library defaults.
  • tls.include_system_ca_certs_pool Type: boolean. Optional. Whether to add certificates from the system certificate authority pool to the configured certificate authorities. If true, the Collector trusts both sources. If false, the Collector trusts only the configured certificate authorities.
  • tls.insecure Type: boolean. Optional. Whether to connect without TLS. If true and no certificate authority is configured, the Collector connects to the brokers without TLS. If false, the Collector connects over TLS.
  • tls.insecure_skip_verify Type: boolean. Optional. Whether to skip verification of the brokers’ certificates. If true, the connection is encrypted but isn’t authenticated and is vulnerable to man-in-the-middle attacks. If false, the Collector verifies the certificates.
  • tls.key_file Type: string. Optional. Path to the PEM-encoded client private key file used with cert_file for mutual TLS authentication.
  • tls.max_version Type: string. Optional. Maximum TLS protocol version the Collector can use. If omitted, the Collector uses the maximum version supported by the Go TLS library.
  • tls.min_version Type: string. Optional. Minimum TLS protocol version the Collector can use. If omitted, the minimum version is TLS 1.2.
  • tls.reload_interval Type: duration. Optional. How often the Collector reloads certificate and key files. If omitted, the Collector doesn’t reload the files.
  • tls.server_name_override Type: string. Optional. Server name to verify in each broker’s certificate instead of the broker’s hostname. Use this setting when the certificates name a host other than the addresses the brokers advertise.
  • metadata Type: object. Optional. How the Collector maintains the cluster metadata it uses to route requests to brokers. Of these settings, only refresh_interval changes what the Collector does.
  • metadata.full Type: boolean. Optional. Default: true. The Collector accepts this setting but ignores it.
  • metadata.refresh_interval Type: duration. Optional. Default: 10m0s. How long the Collector keeps the cluster metadata it caches to route requests to brokers before it refreshes that metadata. Each collection still reads the current list of brokers, topics, and partitions for its metrics. If omitted, the Collector refreshes the cached metadata every 10 minutes.
  • metadata.retry Type: object. Optional. The Collector accepts these settings but ignores them.
  • metadata.retry.backoff Type: duration. Optional. Default: 250ms. The Collector accepts this setting but ignores it.
  • metadata.retry.max Type: integer. Optional. Default: 3. The Collector accepts this setting but ignores it.
  • metrics Type: object. Optional. Per-metric settings, keyed by metric name. Set enabled: true to collect a metric that’s off by default, such as kafka.topic.replication_factor, or enabled: false to stop collecting one. Every metric except kafka.brokers also accepts attributes, the subset of its attributes to keep, and aggregation_strategy, which combines data points that become identical after the other attributes are dropped: sum, avg, min, or max. For example, kafka.consumer_group.lag with attributes: [group, topic] reports one time series per consumer group and topic instead of one per partition. Disabling metrics and dropping attributes also apply to targets discovered through annotations.
  • resource_attributes Type: object. Optional. Which cluster identity resource attributes to report on every metric.
  • resource_attributes.cluster_id Type: object. Optional. Settings for the kafka.cluster.id resource attribute.
  • resource_attributes.cluster_id.enabled Type: boolean. Optional. Default: true. Whether to report the cluster ID from the cluster’s metadata as the kafka.cluster.id resource attribute.
  • resource_attributes.cluster_alias Type: object. Optional. Settings for the kafka.cluster.alias resource attribute.
  • resource_attributes.cluster_alias.enabled Type: boolean. Optional. Default: false. Whether to report cluster_alias as the kafka.cluster.alias resource attribute. Enable it only together with a non-empty cluster_alias.