---
id: event--kafka-producer
title: "Kafka publisher"
summary: "Publishes every incoming record as a message to a Kafka topic. Accepts either a ready-made `EventData` (key, headers, timestamp, binary payload) or a `TextRecord` whose text becomes the payload."
audience: user
keywords: "input, continue, topic, bootstrapservers, username, password, defaultreplicationfactor, groupid, securityprotocol, saslmechanism"
---
# Kafka publisher

Publishes every incoming record as a message to a Kafka topic. Accepts either a ready-made `EventData` (key, headers, timestamp, binary payload) or a `TextRecord` whose text becomes the payload.

## Pins

| Pin | Kind | Types | Description |
|---|---|---|---|
| Input | input | `Viaducta.Plugins.Event.Types.EventData`, `Viaducta.Plugins.TextPlugin.Types.TextRecord` | Record to publish. |
| Continue | continue | same as input | Called with the unchanged input record after the message was handed to the Kafka client. |

## Configuration

| Field | Type | Default | Description |
|---|---|---|---|
| `Topic` | string (max 255) | | Required. Topic to publish to. Created at flow start if it does not exist. |
| `BootstrapServers` | string (max 512) | | Required. Broker address as `host:port`; several brokers comma separated. |
| `Username` | string (max 255) | | SASL user name. See Notes: currently has no effect. |
| `Password` | password (max 255) | | SASL password. See Notes: currently has no effect. |
| `DefaultReplicationFactor` | integer (min 1) | `1` | Replication factor used only when the plugin creates the topic. |
| `GroupId` | string (max 255) | | Not used yet: the consumer always uses a random group id (see Notes). |
| `SecurityProtocol` | dropdown: `0` Plaintext, `1` Ssl, `2` SaslPlaintext, `3` SaslSsl | `0` | Not used yet: the client always connects in plaintext (see Notes). |
| `SaslMechanism` | dropdown: `1` Plain, `2` ScramSha256, `3` ScramSha512 | `1` | Visible with `SecurityProtocol` `2` / `3`. Not used yet. |

## Behaviour

- At flow start the producer is created and the topic is checked (metadata request with a 1 second timeout) and created with `DefaultReplicationFactor` if missing; a creation error is only logged. The plugin keeps the flow running (KeepAlive).
- Message built from the input:
  - `EventData`: `Id` → message key, `Data` → value, `Headers` → message headers (values encoded as UTF-8, null becomes empty), `Time` → message timestamp (current UTC time if null).
  - `TextRecord`: `Value` → value (UTF-8 bytes; null value gives an empty message), key = new random GUID (32 hex characters), timestamp = now, no headers. `Name` and `ContentType` are not sent.
  - Any other type raises an error.
- The message is queued in the Kafka client (fire-and-forget) and `Continue` is called immediately with the input record.

## Notes

- **No delivery confirmation:** `Continue` is called before the broker confirmed the message and delivery errors are not reported to the flow. Messages still queued when the flow stops can be lost.
- **Authentication:** `Username` / `Password` are set as SASL credentials, but the security protocol and SASL mechanism are not set, so the client connects in plaintext without authentication. TLS is not supported.
- The topic is fixed per component; the record cannot choose the topic.
- To control the key or add headers, build an `EventData` (for example in a script) instead of sending a `TextRecord`.

## Example

```json
{ "Topic": "orders", "BootstrapServers": "kafka1:9092", "DefaultReplicationFactor": 1 }
```

Http listener → Json serializer (`TextRecord`) → **Kafka publisher** → Http response