Kafka

The Apache Kafka integration requires CXDOT Collector 1.7.0 or greater. Apache Kafka is an open source distributed event streaming platform. Use the Kafka integration to collect broker and Java Virtual Machine (JVM) runtime metrics from the Kafka brokers running in your environment. The Kafka integration supports Apache Kafka 3.8 or greater.

​

Supported telemetry types

The Kafka integration supports these telemetry types:

​

Prerequisites

The Kafka integration has the following prerequisites:

• Enable remote Java Management Extensions (JMX) monitoring on each broker. Kafka disables remote JMX by default. Set the JMX_PORT environment variable before you start the broker, or set the equivalent com.sun.management.jmxremote system properties in KAFKA_JMX_OPTS. JMX_PORT sets the same port for the JMX registry and for the connection it returns, so the broker needs one open port for JMX. For more information, see Monitoring in the Kafka documentation.
• Make the JMX port and the address the broker advertises for JMX connections reachable from the Collector. Set java.rmi.server.hostname in KAFKA_JMX_OPTS to an address the Collector can resolve, or configure the integration’s advertised host and port overrides when that address isn’t reachable. For example, the following environment variables enable remote JMX on port 9999 without authentication or Transport Layer Security (TLS):
  export JMX_PORT=9999
  export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true \
    -Dcom.sun.management.jmxremote.authenticate=false \
    -Dcom.sun.management.jmxremote.ssl=false \
    -Djava.rmi.server.hostname=BROKER_ADDRESS"
  
  Replace BROKER_ADDRESS with the broker’s address as the Collector reaches it. In production, enable JMX authentication and TLS on the broker and configure the integration to match.

​

Configure

The Kafka integration is enabled by default. To configure the Kafka integration, follow these steps:

1. For Kafka brokers running in your Kubernetes cluster, add autodiscovery annotations for Kafka to the broker pods or their Service, and supply the JMX address and any JMX credentials in the annotation. For more information, see autodiscovery. Metrics from discovered pods arrive with that pod’s Kubernetes metadata attached.
2. Optional: Configure static targets instead of discovered targets, such as brokers outside your Kubernetes cluster. For example, add the following to the values.yaml for your CXDOT Collector Helm chart:
   config:
     integrations:
       kafka:
         endpoints:
           - endpoint: kafka-0.example.com:9999
   
   When you configure static targets, this integration instance collects from exactly those targets instead of discovered targets.
3. Optional: Configure a second integration instance to collect from both discovered brokers and static targets. For example, add the following to the values.yaml for your CXDOT Collector Helm chart:
   config:
     integrations:
       kafka: {}
       kafka/edge:
         endpoints:
           - endpoint: kafka-0.example.com:9999
   
   The kafka/edge key defines a named instance for the static targets. The bare kafka key continues to discover targets.
4. Optional: If the brokers require JMX authentication or TLS, configure the matching settings for static targets, and provide the password through your deployment’s secret management. For example, the following Helm values reference the KAFKA_JMX_PASSWORD environment variable:
   config:
     integrations:
       kafka:
         endpoints:
           - endpoint: kafka-0.example.com:9999
         username: cxdot
         password: "${env:KAFKA_JMX_PASSWORD}"
   
   For brokers that autodiscovery finds, supply the credentials in the autodiscovery annotation instead.
5. Optional: Disable the integration. For example, add the following to the values.yaml for your CXDOT Collector Helm chart:
   config:
     integrations:
       kafka:
         enabled: false
   

​

Validate

To validate the Kafka integration, follow these steps:

1. In Live Telemetry Analyzer, filter for __name__=cxdot.integration.target.health and cxdot.integration.name=kafka. Confirm that each broker reports 1. The cxdot.integration.target, server.address, and server.port attributes identify the broker.
2. In Metrics Explorer, run the following query:
   sum by ("server.address", "server.port") ({"kafka.broker.partition.count"})
   
   Confirm that the query returns one time series per broker, with that broker’s partition count as the value.

​

Troubleshooting

• Metrics for a new topic, or for a request type the brokers haven’t served yet, are missing: Kafka registers the beans for topics and request types when they’re first used, and the integration refreshes its list of beans at each bean_detection_interval, which defaults to 10 minutes. The missing time series appear after the next refresh.
• The kafka.broker.raft.*, kafka.broker.metadata.*, and kafka.broker.snapshot.* metrics are missing: Only a broker running in KRaft mode reports them, except for kafka.broker.metadata.controller, which brokers in both modes report. A broker running in ZooKeeper mode reports kafka.broker.zookeeper.session.event.count and kafka.broker.leader.election.count instead, which a KRaft broker never reports.

​

Configuration reference

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

​

Optional settings

• enabled Type: boolean. Optional. Default: true. Whether to enable this Kafka 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 broker JMX targets. When this list is empty or omitted, the integration uses automatic target discovery. When the list contains targets, this integration instance collects from exactly those targets and doesn’t automatically discover targets. Configure a second named instance to use both methods. Each endpoint can be a host:port RMI registry address or a service:jmx:rmi URL.
• endpoints[].endpoint Type: string. Required. Kafka broker JMX endpoint as either a host:port RMI registry address or a service:jmx:rmi URL.
• registry_tls Type: boolean. Optional. Default: false. Whether to use TLS for the RMI registry connection. If true, the Collector uses TLS. If false, the Collector uses an unencrypted registry connection.
• advertised_host_override Type: string. Optional. Default: “. Host the Collector uses for Remote Method Invocation (RMI) server connections instead of the host advertised by returned RMI stubs. When empty, the Collector uses the registry endpoint host.
• advertised_port_override Type: integer. Optional. Default: 0. Minimum: 0. Maximum: 65535. Port the Collector uses for RMI server connections instead of the port advertised by returned RMI stubs. When set to 0, the Collector uses the advertised port.
• username Type: string. Optional. Username for JMX authentication. Configure username and password together.
• password Type: string. Optional. Password for JMX authentication. Configure username and password together.
• collection_interval Type: duration. Optional. Default: 15s. How often the Collector collects metrics from each Kafka broker JMX endpoint.
• bean_detection_interval Type: duration. Optional. Default: 10m. How often the Collector refreshes the set of available JMX beans. Kafka registers beans for topics and request types when they’re first used, so their metrics appear after the next refresh.
• bean_limit Type: integer. Optional. Default: 35000. Minimum: 1. Maximum number of JMX bean matches collected from each endpoint. When the number of matches exceeds this value, the integration drops beans according to the attribute mapping priority.
• target_system Type: string. Optional. Default: kafka. Comma-delimited attribute mapping names to apply. The Kafka broker mapping is enabled by default, and the JVM mapping is included automatically. Add -jvm to disable JVM metrics when at least one other mapping is configured.
• attribute_mapping_paths Type: array of string. Optional. Default: []. Directories and files containing custom Java Management Extensions (JMX) attribute mappings. The Collector searches these paths before the embedded Kafka and Java Virtual Machine (JVM) mappings.
• tls Type: object. Optional. Default: {}. TLS certificate and client authentication settings for RMI registry and server connections.
• tls.ca_file Type: string. Optional. Path to the CA cert. For a client this verifies the server certificate. For a server this verifies client certificates. If empty uses system root CA. (optional)
• tls.ca_pem Type: string. Optional. In memory PEM encoded cert. (optional)
• tls.cert_file Type: string. Optional. Path to the TLS cert to use for TLS required connections. (optional)
• tls.cert_pem Type: string. Optional. In memory PEM encoded TLS cert to use for TLS required connections. (optional)
• tls.cipher_suites Type: array of string. Optional. CipherSuites is a list of TLS cipher suites that the TLS transport can use. If left blank, a safe default list is used. See https://go.dev/src/crypto/tls/cipher_suites.go for a list of supported cipher suites.
• tls.curve_preferences Type: array of string. Optional. contains the elliptic curves that will be used in an ECDHE handshake, in preference order Defaults to empty list and “crypto/tls” defaults are used, internally.
• tls.include_system_ca_certs_pool Type: boolean. Optional. If true, load system CA certificates pool in addition to the certificates configured in this struct.
• tls.insecure Type: boolean. Optional. Whether to disable transport security. This setting must be false. Use registry_tls to control transport security for the RMI registry.
• tls.insecure_skip_verify Type: boolean. Optional. Whether to skip verification of the JMX server certificate. 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 certificate.
• tls.key_file Type: string. Optional. Path to the TLS key to use for TLS required connections. (optional)
• tls.key_pem Type: string. Optional. In memory PEM encoded TLS key to use for TLS required connections. (optional)
• tls.max_version Type: string. Optional. MaxVersion sets the maximum TLS version that is acceptable. If not set, refer to crypto/tls for defaults. (optional)
• tls.min_version Type: string. Optional. MinVersion sets the minimum TLS version that is acceptable. If not set, TLS 1.2 will be used. (optional)
• tls.reload_interval Type: duration. Optional. ReloadInterval specifies the duration after which the certificate will be reloaded If not set, it will never be reloaded (optional)
• tls.server_name_override Type: string. Optional. Server name to verify in the JMX certificate instead of the endpoint hostname. Use this setting when the certificate name differs from the endpoint hostname.
• tls.tpm Type: object. Optional. Trusted platform module configuration
• tls.tpm.auth Type: string. Optional. Authorization value for the trusted platform module key.
• 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.owner_auth Type: string. Optional. Owner authorization value for the trusted platform module.
• tls.tpm.path Type: string. Optional. The path to the TPM device or Unix domain socket. For instance /dev/tpm0 or /dev/tpmrm0.