> For the complete documentation index, see [llms.txt](https://docs.onum.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.onum.com/data-sinks/data-sink-integrations/send-data-to-apache-kafka.md).

# Send data to Apache Kafka

{% hint style="info" %}
See the changelog of the **Apache Kafka** Data Sink [here](/data-sinks-changelog/apache-kafka-data-sink.md).
{% endhint %}

## Overview

Onum supports integration with [Apache Kafka](https://kafka.apache.org/).

The **Apache Kafka** Data sink lets you send data messages to a Kafka topic, using information from events to construct messages. This component supports various configurations, including Kafka brokers, topic partitions, authentication methods, and message compression, providing a flexible and scalable approach for integrating with Kafka.

## Prerequisites 

You will need to set up a running Kafka cluster, with optional group IDs and Topics.

## Onum Setup

{% stepper %}
{% step %}
Log in to your Onum tenant and click **Data Sinks > New Data sink**.
{% endstep %}

{% step %}
Double-click the **Apache Kafka** Data Sink.
{% endstep %}

{% step %}
Enter a **Name** for the new Data Sink. Optionally, add a **Description** and some **Tags** to identify the Data Sink.
{% endstep %}

{% step %}
Decide whether or not to include this Data Sink info in the metrics and graphs of the [**Home**](/home.md) area.
{% endstep %}

{% step %}
**Kafka Brokers**

The broker hosts the partitions and handles incoming requests. Find this in the **Instances** area. The format is `Kafka installed Machine HostName:Port Number.`

Add the corresponding **Host** and **Port** and click **Add element** to add the required number of brokers.
{% endstep %}

{% step %}
In the **Authentication** section, you may enter your authentication parameters (either **SASL** or **SSL/TLS**). Otherwise, select **None**.

<details>

<summary><strong>SSL/TLS</strong></summary>

* **Minimum TLS version** - Choose your minimum TLS version.
* **Certificate** - Add your TLS certificate from your [Secrets](/settings/organization-settings/secrets-management.md) or create one.
* **Private Key** - Add your private key from your [Secrets](/settings/organization-settings/secrets-management.md) or create one.
* **Skip TLS Verification** - Toggle on to skip TLS verification.
* **CA Chain** - Add your CA chain from your [Secrets](/settings/organization-settings/secrets-management.md) or create one.

{% hint style="info" %}
Learn more about secrets in Onum in [this article](/settings/organization-settings/secrets-management.md).
{% endhint %}

</details>

<details>

<summary><strong>SASL</strong></summary>

* **Mechanism** - SASL provides different mechanisms (methods) for authentication. Choose between **PLAIN**, **SCRAM-SHA-256**, **SCRAM-SHA-512**, **OAUTHBEARER**, and **GSSAPI**.
* **saslUsername** - Enter your SASL name.
* **saslPassword** - Add your password from your [Secrets](/settings/organization-settings/secrets-management.md) or create one.

{% hint style="info" %}
Learn more about secrets in Onum in [this article](/settings/organization-settings/secrets-management.md).
{% endhint %}

</details>
{% endstep %}

{% step %}
Configure your **Topic & Partition:**

* **Kafka topic -** Use `kafka-topics --broker :9092 --describe` and write the result here.
* The **Partition** is the subdivision of the topic, designed to allow parallelism, scalability, and fault tolerance in message processing. This is where it is contained within the specified broker. Enter `-1` for automatic selection.
  * &#x20;`0` - \[msg1, msg4, msg7]
  * &#x20;`1` **-** \[msg2, msg5, msg8]
    {% endstep %}

{% step %}
The **Acknowledgment** (acks) setting controls how producers handle message durability and delivery confirmation from brokers. It defines how many replicas must confirm the write before the producer considers it successful.&#x20;

* **All** - Full replication. The leader waits for all in-sync replicas (ISR) to acknowledge before confirming.&#x20;
* **-1** - This is the same as **All**.
* **0** - No acknowledgmen&#x74;**.** The producer sends data but does **not wait** for confirmation.
* **1** - Leader acknowledgment. The leader broker confirms receipt **without waiting** for replicas.
  {% endstep %}

{% step %}
In the **Advanced configuration** section, you can toggle on the **Batch configuration** switch button to enable batch settings. The values will be filled in by default; however, you can change them if needed.

<table data-search="false"><thead><tr><th width="169.2265625">Parameter</th><th>Description</th></tr></thead><tbody><tr><td><strong>Batch size</strong> </td><td>The batch size (in bytes) to allow through. The default value is <code>1000000</code>.</td></tr><tr><td><strong>Batch number of messages</strong></td><td>The number of messages to allow through. The default value is <code>10000</code>.</td></tr><tr><td><strong>Queue max number of messages</strong></td><td>Enter how many events to collect until considered full. The default value is <code>100000</code>.</td></tr><tr><td><strong>Queue max size</strong></td><td>Enter the maximum number of events in KB to collect until considered full. The default value is <code>102400</code>.</td></tr><tr><td><strong>Linger</strong></td><td>Enter the number of seconds to wait before considering the batch completed. The default value is <code>5</code>.</td></tr><tr><td><strong>Max in-flight requests per connection</strong></td><td>Controls how many unacknowledged produce requests can be sent at the same time on a single TCP connection between a Kafka producer and a broker. The default value is <code>1000000</code>.</td></tr><tr><td><strong>Delivery timeout</strong></td><td>The maximum amount of time to wait for a message to be successfully sent before it is marked as failed. The default value is <code>300000</code>.</td></tr><tr><td><strong>Request timeout</strong></td><td>The number of milliseconds to wait before considering the batch request a timeout. The default value is <code>30000</code>.</td></tr></tbody></table>
{% endstep %}

{% step %}
Toggle on the **Use compression** switch if you need to compress your data and choose the required type between **Gzip**, **Snappy**, **LZ4**, or **Zstd**.
{% endstep %}

{% step %}
Enter the **Number of times** to retry the server before deeming the request as timed out. The default value is `3`, and the minimum value is `0`.
{% endstep %}

{% step %}
Click **Finish** when complete. Your new Data Sink will appear in the **Data sinks** area list.
{% endstep %}
{% endstepper %}

## Pipeline configuration

When it comes to using this Data sink in a [Pipeline](/pipelines.md), you must configure the following output parameters. To do it, simply click the Data sink on the canvas and select **Configuration**.

### Output configuration

<table><thead><tr><th width="170.01953125">Parameter</th><th>Description</th></tr></thead><tbody><tr><td><strong>Message</strong></td><td>Choose the field containing the message to send.</td></tr></tbody></table>

#### Advanced configuration

<table><thead><tr><th width="170.03515625">Parameter</th><th>Description</th></tr></thead><tbody><tr><td><strong>Message key</strong></td><td>If the message has a key, choose the field containing it here.</td></tr></tbody></table>

Click **Save** to save your configuration.