# Kafka

Writes each record to a Kafka topic.

Smeltr manages the producer, including its protection against duplicates caused by producer retries.

## Example

Writes to a plaintext test broker. Replace its address with one reachable from the worker, and keep input and output topics separate unless you mean to build a feedback path.

```
{
  "id": "events-out",
  "type": "kafka",
  "brokers": "localhost:9092",
  "defaultTopic": "smeltr-events-out"
}
```

## Before you start

-   Create the output topic.
-   Allow workers to reach every broker address Kafka advertises.
-   Give the Kafka account permission to write to the topic.
-   For an authenticated broker, configure [Kafka security](/docs/secrets#kafka). Select a [connection identity](/docs/secrets#identities) for a private CA or a client certificate.

## Settings

It also takes the settings [every destination](/docs/destinations#settings) takes.

`brokers`  string required

The brokers a worker first connects to, `host:port`, separated by commas.

Examples:

- `kafka-1:9092,kafka-2:9092`
- `broker.example.com:9093`

`defaultTopic`  string required

The topic for a record with no `__destination.topic`.

Examples:

- `orders-clean`
- `security.firewall.v1`

`format`  string default `json`

What each message holds.

`json`

Writes the record as JSON.

`text`

Writes one field's text, from `field`.

`bytes`

Writes each source record's bytes as read, before pipeline processing, once per record: `__source.bytes` unless `field` names another `bytes` field.

`avro`

Writes Avro messages, registering the schema in the schema registry.

`schemaRegistry`  object default none

The schema registry the Avro schema is registered in. Set `url`. If the registry needs a sign-in, also set `username` and `password`. Needed for format `avro`.

Examples:

- `{"url": "https://registry.example.com"}`
- `{"url": "https://registry.example.com","username": "smeltr","password": {"secret": "registry","version": 1}}`

`field`  string default `_raw` for `text`, `__source.bytes` for `bytes`

The field each `text` or `bytes` record writes. Only with format `text` or `bytes`.

Example:

- `message`

`identity`  identity default none

The [connection identity](/docs/secrets#identities) it connects under, for a private CA or a client certificate. Without one, no certificate is offered and the public roots, Mozilla's and the host's, are trusted.

Examples:

- `kafka-client`
- `kafka-mtls`

### Advanced: Delivery and client properties

`compression`  string default `none`

How it compresses each batch of messages.

`none`

Sends it uncompressed.

`gzip`

Compresses it with gzip.

`snappy`

Compresses it with Snappy.

`lz4`

Compresses it with LZ4.

`zstd`

Compresses it with Zstandard.

`stallAfterSecs`  integer default `30`

Seconds one batch may take to deliver before it counts as stalled. 1 to 120.

Examples:

- `60`
- `120`

`bufferBytes`  size default 64 MiB

How much undelivered output it may hold. The Kafka client holds as much again. 4 MiB to 1 GiB.

Examples:

- `256 MiB`
- `16 MiB`
- `1 GiB`

`properties`  map default none

Other Kafka client properties, each name to its value. See [Kafka security](/docs/secrets#kafka). A property Smeltr sets itself, such as `bootstrap.servers`, is refused, and a secret one belongs in `credentials`.

Examples:

- `security.protocol = SASL_SSL`
- `security.protocol = SASL_SSL`
- `sasl.mechanism = SCRAM-SHA-512`
- `sasl.username = smeltr`

`credentials`  map default none

Kafka client properties whose values are secret, each name to the [credential](/docs/secrets) holding its value.

Examples:

- `sasl.password = KAFKA_PASSWORD`
- `sasl.oauthbearer.client.secret = KAFKA_CLIENT_SECRET`

## Fields it reads

| Field | Type | Holds |
| --- | --- | --- |
| `__destination.topic` | `str` | The Kafka topic or Event Hub a destination writes the record to. Null means the default value for that destination. |

## Verify

Send a record you can recognize through the source, consume the output topic, and check the record's fields, not just that the topic has messages. The [tutorial's delivery check](/docs/first-pipeline#verify) shows the commands.

## Troubleshooting

**Timeout or authorization error**

Check advertised addresses, that the topic exists, write permissions and [Kafka security](/docs/secrets#kafka).

## See also

-   [Destinations](/docs/destinations)
-   [Kafka source](/docs/sources/kafka)
-   [Kafka security](/docs/secrets#kafka)
-   [Handle refused records](/docs/delivery#refused)