> ## Documentation Index
> Fetch the complete documentation index at: https://docs-xcor.paloaltonetworks.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Kafka

> Apache Kafka broker and Java Virtual Machine runtime metrics collected over JMX.

The Apache Kafka integration requires CXDOT Collector 1.7.0 or greater.

[Apache Kafka](https://kafka.apache.org/) 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:

| Type | Supported |
| :-: | :-: |
| Logs | No |
| Metrics | Yes |
| Traces | No |
| Events | No |

## 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](https://kafka.apache.org/documentation/#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):

  ```shell theme={null}
  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](https://docs-xcor.paloaltonetworks.com/ingest/xcor/collector/discover/kubernetes).

   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:

   ```yaml theme={null}
   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:

   ```yaml theme={null}
   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:

   ```yaml theme={null}
   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:

   ```yaml theme={null}
   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:

   ```text theme={null}
   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](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.


## Related topics

- [Kafka destination plugin](/ingest/pipeline/plugins/destination-plugins/kafka.md)
- [Kafka source plugin](/ingest/pipeline/plugins/source-plugins/kafka.md)
- [Azure metrics available in Palo Alto Networks Cortex XCOR](/ingest/xcor/integrations/server/azure/azure-metrics.md)
- [Redpanda Cloud destination plugin](/ingest/pipeline/plugins/destination-plugins/redpanda.md)
- [Confluent Cloud DataSet destination plugin](/ingest/pipeline/plugins/destination-plugins/confluent-cloud.md)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.