---
title: Configure the Kafka output
description: The Kafka output sends events to Apache Kafka. To use this output, edit the Auditbeat configuration file to deactivate the Elasticsearch output by commenting...
url: https://www.elastic.co/docs/reference/beats/auditbeat/kafka-output
products:
  - Auditbeat
  - Beats
applies_to:
  - Elastic Cloud Serverless: Generally available
  - Elastic Stack: Generally available
---

# Configure the Kafka output
The Kafka output sends events to Apache Kafka.
To use this output, edit the Auditbeat configuration file to deactivate the Elasticsearch output by commenting it out, and enable the Kafka output by uncommenting the Kafka section.
<admonition title="Kafka timestamps and beats">
  - Kafka 3.6+ introduces stricter timestamp validation with the introduction of two new broker/topic-level properties: [log.message.timestamp.before.max.ms](https://docs.confluent.io/platform/current/installation/configuration/topic-configs.html#message-timestamp-before-max-ms) and
    [log.message.timestamp.after.max.ms](https://docs.confluent.io/platform/current/installation/configuration/topic-configs.html#message-timestamp-after-max-ms).
    These properties limit the time difference between the message timestamp (from beats) and the Kafka broker receive time.
    Messages can be rejected if the values are exceeded and `log.message.timestamp.type=CreateTime` is set.
    These checks are ignored if `log.message.timestamp.type=LogAppendTime` is set.
  - For Kafka version 0.10.0.0+ the message creation timestamp is set by beats and equals the initial timestamp of the event. This behavior affects the retention policy in Kafka. For example, if a beat event was created 2 weeks ago, the retention policy is set to 7 days and the message from beats arrives to Kafka today, it is immediately discarded because the timestamp value is before the last 7 days.
    You can change this behavior by setting timestamps on message arrival instead.
    The message is not discarded but kept for 7 more days. Set `log.message.timestamp.type` to `LogAppendTime` (default `CreateTime`) in the Kafka configuration.
</admonition>

Example configuration:
```yaml
output.kafka:
  # initial brokers for reading cluster metadata
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]

  # message topic selection + partitioning
  topic: '%{[fields.log_topic]}'
  partition.round_robin:
    reachable_only: false

  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000
```

<note>
  Events bigger than [`max_message_bytes`](#kafka-max_message_bytes) will be dropped. To avoid this problem, make sure Auditbeat does not generate events bigger than [`max_message_bytes`](#kafka-max_message_bytes).
</note>


## Compatibility

This output can connect to Kafka version 0.8.2.0 and later. Older versions might work as well, but are not supported. When using Kafka 4.0 and newer, the version must be set to at least `"2.1.0"`

## Configuration options

You can specify the following options in the `kafka` section of the `auditbeat.yml` config file:

### `enabled`

The `enabled` config is a boolean setting to enable or disable the output. If set to false, the output is disabled.
The default value is `true`.

### `hosts`

The list of Kafka broker addresses from where to fetch the cluster metadata. The cluster metadata contain the actual Kafka brokers events are published to.

### `version`

Kafka protocol version that Auditbeat will request when connecting. Defaults to 2.1.0. When using Kafka 4.0 and newer, the version must be set to at least `"2.1.0"`
Valid values are all kafka releases in between `0.8.2.0` and `4.1.0`.
The protocol version controls the Kafka client features available to Auditbeat; it does not prevent Auditbeat from connecting to Kafka versions newer than the protocol version.
See [Compatibility](#kafka-compatibility) for information on supported versions.

### `username`

The username for connecting to Kafka. If username is configured, the password must be configured as well.

### `password`

The password for connecting to Kafka.

### `sasl.mechanism`

The SASL mechanism to use when connecting to Kafka. It can be one of:
- `PLAIN` for SASL/PLAIN.
- `SCRAM-SHA-256` for SCRAM-SHA-256.
- `SCRAM-SHA-512` for SCRAM-SHA-512.

If `sasl.mechanism` is not set, `PLAIN` is used if `username` and `password` are provided. Otherwise, SASL authentication is disabled.
To use `GSSAPI` mechanism to authenticate with Kerberos, you must leave this field empty, and use the [`kerberos`](#kerberos-option-kafka) options.

### `topic`

The Kafka topic used for produced events.
You can set a static topic, for example `auditbeat`, or you can use a format string to set a topic dynamically based on one or more [Elastic Common Schema (ECS)](https://www.elastic.co/docs/reference/ecs) fields. Available fields include:
- `data_stream.type`
- `data_stream.dataset`
- `data_stream.namespace`
- `@timestamp`
- `event.dataset`

For example:
```yaml
topic: '%{[data_stream.type]}-%{[data_stream.dataset]}-%{[data_stream.namespace]}'
```

You can also set a custom field. This is useful if you need to construct a more complex or structured topic name. For example, this configuration uses the `fields.log_topic` custom field to set the topic for each event:
```yaml
topic: '%{[fields.log_topic]}'
```

<tip>
  To learn how to add custom fields to events, see the [`fields`](/docs/reference/beats/auditbeat/configuration-general-options#libbeat-configuration-fields) option.
</tip>

To set a dynamic topic value for outputting Auditbeat data to Kafka, you can add the [`add_fields` processor](https://www.elastic.co/docs/reference/beats/auditbeat/add-fields) to Auditbeat's input configuration settings.
For example, the following `add_fields` processor creates a dynamic topic value for the `fields.log_topic` field by combining multiple [ECS data stream fields](https://www.elastic.co/docs/reference/ecs/ecs-data_stream):
```yaml
- add_fields:
    target: ''
    fields:
      log_topic: '%{[data_stream.type]}-%{[data_stream.dataset]}-%{[data_stream.namespace]}' 
```

For more information, refer to [Filter and enhance data with processors](https://www.elastic.co/docs/reference/beats/auditbeat/filtering-enhancing-data).
See the [`topics`](#topics-option-kafka) setting for other ways to set the topic dynamically.

### `topics`

An array of topic selector rules. Each rule specifies the `topic` to use for events that match the rule. During publishing, Auditbeat sets the `topic` for each event based on the first matching rule in the array. Rules can contain conditionals, format string-based fields, and name mappings. If the `topics` setting is missing or no rule matches, the [`topic`](#topic-option-kafka) field is used.
Rule settings:
<definitions>
  <definition term="topic">
    The topic format string to use.  If this string contains field references, such as `%{[fields.name]}`, the fields must exist, or the rule fails.
  </definition>
  <definition term="mappings">
    A dictionary that takes the value returned by `topic` and maps it to a new name.
  </definition>
  <definition term="default">
    The default string value to use if `mappings` does not find a match.
  </definition>
  <definition term="when">
    A condition that must succeed in order to execute the current rule. All the [conditions](/docs/reference/beats/auditbeat/defining-processors#conditions) supported by processors are also supported here.
  </definition>
</definitions>

The following example sets the topic based on whether the message field contains the specified string:
```yaml
output.kafka:
  hosts: ["localhost:9092"]
  topic: "logs-%{[agent.version]}"
  topics:
    - topic: "critical-%{[agent.version]}"
      when.contains:
        message: "CRITICAL"
    - topic: "error-%{[agent.version]}"
      when.contains:
        message: "ERR"
```

This configuration results in topics named `critical-[version]`, `error-[version]`, and `logs-[version]`.

### `key`

Optional formatted string specifying the Kafka event key. If configured, the event key can be extracted from the event using a format string.
See the Kafka documentation for the implications of a particular choice of key; by default, the key is chosen by the Kafka cluster.

### `partition`

Kafka output broker event partitioning strategy. Must be one of `random`, `round_robin`, or `hash`. By default the `hash` partitioner is used.
<definitions>
  <definition term="random.group_events">
    Sets the number of events to be published to the same partition, before the partitioner selects a new partition by random. The default value is 1 meaning after each event a new partition is picked randomly.
  </definition>
  <definition term="round_robin.group_events">
    Sets the number of events to be published to the same partition, before the partitioner selects the next partition. The default value is 1 meaning after each event the next partition will be selected.
  </definition>
  <definition term="hash.hash">
    List of fields used to compute the partitioning hash value from. If no field is configured, the events `key` value will be used.
  </definition>
  <definition term="hash.random">
    Randomly distribute events if no hash or key value can be computed.
  </definition>
</definitions>

All partitioners will try to publish events to all partitions by default. If a partition’s leader becomes unreachable for the beat, the output might block. All partitioners support setting `reachable_only` to overwrite this behavior. If `reachable_only` is set to `true`, events will be published to available partitions only.
<note>
  Publishing to a subset of available partitions potentially increases resource usage because events may become unevenly distributed.
</note>


### `headers`

A header is a key-value pair, and multiple headers can be included with the same `key`. Only string values are supported. These headers will be included in each produced Kafka message.
```yaml
output.kafka:
  hosts: ["localhost:9092"]
  topic: "logs-%{[agent.version]}"
  headers:
    - key: "some-key"
      value: "some value"
    - key: "another-key"
      value: "another value"
```


### `client_id`

The configurable ClientID used for logging, debugging, and auditing purposes. The default is "beats".

### `codec`

Output codec configuration. If the `codec` section is missing, events will be json encoded.
See [Change the output codec](https://www.elastic.co/docs/reference/beats/auditbeat/configuration-output-codec) for more information.

### `metadata`

Kafka metadata update settings. The metadata do contain information about brokers, topics, partition, and active leaders to use for publishing.
<definitions>
  <definition term="refresh_frequency">
    Metadata refresh interval. Defaults to 10 minutes.
  </definition>
  <definition term="full">
    Strategy to use when fetching metadata, when this option is `true`, the client will maintain a full set of metadata for all the available topics, if the this option is set to `false` it will only refresh the metadata for the configured topics. The default is false.
  </definition>
  <definition term="retry.max">
    Total number of metadata update retries when cluster is in middle of leader election. The default is 3.
  </definition>
  <definition term="retry.backoff">
    Waiting time between retries during leader elections. Default is 250ms.
  </definition>
</definitions>


### `max_retries`

The number of times to retry publishing an event after a publishing failure. After the specified number of retries, the events are typically dropped.
Set `max_retries` to a value less than 0 to retry until all events are published.
The default is 3.

### `backoff.init`

The number of seconds to wait before trying to republish to Kafka after a network error. After waiting `backoff.init` seconds, Auditbeat tries to republish. If the attempt fails, the backoff timer is increased exponentially up to `backoff.max`. After a successful publish, the backoff timer is reset. The default is 1s.

### `backoff.max`

The maximum number of seconds to wait before attempting to republish to Kafka after a network error. The default is 60s.

### `bulk_max_size`

The maximum number of events to bulk in a single Kafka request. The default is 2048.

### `bulk_flush_frequency`

Duration to wait before sending bulk Kafka request. 0 is no delay. The default is 0.

### `timeout`

The number of seconds to wait for responses from the Kafka brokers before timing out. The default is 30 (seconds).

### `broker_timeout`

The maximum duration a broker will wait for number of required ACKs. The default is 10s.

### `channel_buffer_size`

Per Kafka broker number of messages buffered in output pipeline. The default is 256.

### `keep_alive`

The keep-alive period for an active network connection. If 0s, keep-alives are disabled. The default is 0 seconds.

### `compression`

Sets the output compression codec. Must be one of `none`, `snappy`, `lz4`, `gzip` and `zstd`. The default is `gzip`.
<admonition title="Known issue with Azure Event Hub for Kafka">
  When targeting Azure Event Hub for Kafka, set `compression` to `none` as the provided codecs are not supported.
</admonition>


### `compression_level`

Sets the compression level used by gzip. Setting this value to 0 disables compression. The compression level must be in the range of 1 (best speed) to 9 (best compression).
Increasing the compression level will reduce the network usage but will increase the cpu usage.
The default value is 4.

### `max_message_bytes`

The maximum permitted size of JSON-encoded messages. Bigger messages will be dropped. The default value is 1000000 (bytes). This value should be equal to or less than the broker’s `message.max.bytes`.

### `required_acks`

The ACK reliability level required from broker. 0=no response, 1=wait for local commit, -1=wait for all replicas to commit. The default is 1.
Note: If set to 0, no ACKs are returned by Kafka. Messages might be lost silently on error.

### `ssl`

Configuration options for SSL parameters like the root CA for Kafka connections. The Kafka host keystore should be created with the `-keyalg RSA` argument to ensure it uses a cipher supported by [Filebeat’s Kafka library](https://github.com/Shopify/sarama/wiki/Frequently-Asked-Questions#why-cant-sarama-connect-to-my-kafka-cluster-using-ssl). See [SSL](https://www.elastic.co/docs/reference/beats/auditbeat/configuration-ssl) for more information.

### `kerberos`

<warning>
  This functionality is in beta and is subject to change. The design and code is less mature than official GA features and is being provided as-is with no warranties. Beta features are not subject to the support SLA of official GA features.
</warning>

Configuration options for Kerberos authentication.
See [Kerberos](https://www.elastic.co/docs/reference/beats/auditbeat/configuration-kerberos) for more information.

### `queue`

Configuration options for internal queue.
See [Internal queue](https://www.elastic.co/docs/reference/beats/auditbeat/configuring-internal-queue) for more information.
Note:`queue` options can be set under `auditbeat.yml` or the `output` section but not both.