<!-- BusyBot node reference — https://busybot.net/tools/kafka/ -->

> Node: Kafka (`kafka`) · Action (binary) · v1
> Category: Development · Credentials: Kafka (`kafkaApi`)
> Updated: 2026-08-16

# Kafka

> Produce messages to Apache Kafka topics

## Overview

The Kafka tool produces messages to Apache Kafka topics using the kafkajs client library. It supports sending item JSON as message value, custom string messages, or binary data from upstream items. Features include: Confluent Schema Registry encoding, message keys, custom headers (UI or JSON), GZIP compression, and ack configuration. Messages are batched per topic and sent via producer.sendBatch() for efficiency. Authentication supports SASL (plain, scram-sha-256, scram-sha-512) with optional SSL.

**Category:** Development  
**Tool Name:** `kafka`  
**Version:** 1

**Appearance:** Icon: `si-apachekafka` | Color: `#231F20`

## Node Type

**Action (Binary)** — handles file/binary data operations

## Input / Output

| Direction | Port(s) |
|-----------|--------|
| Input | `Input` |
| Output | `Output`, `Error` |

## Credentials

This tool requires **Kafka** credentials.
See the [Credentials Guide](https://busybot.net/credentials/kafka-api/) for setup instructions.

### Parameters

| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| Topic | `string` | Yes | — | Name of the Kafka topic to publish to. |
| Send Input Data | `boolean` | No | `true` | Whether to send the incoming item JSON as the Kafka message value. |
| Message | `string` | No | — | Custom message string to send (when Send Input Data is false). _(shown when Send Input Data is `false`)_ |
| Use Binary Data | `boolean` | No | `false` | Whether to read binary data from the item and send it as the message value. _(shown when Send Input Data is `false`)_ |
| Binary Property | `string` | No | `data` | Name of the binary property to read the message value from. Names are case-sensitive — see the upstream node's Binary Data panel for the exact names to use. _(shown when Send Input Data is `false` and Use Binary Data is `true`)_ |
| Use Schema Registry | `boolean` | No | `false` | Whether to encode messages via Confluent Schema Registry. |
| Schema Registry URL | `string` | Yes | — | URL of the Confluent Schema Registry. _(shown when Use Schema Registry is `true`)_ |
| Event Name | `string` | Yes | — | Namespace and name of the schema in the Schema Registry (namespace.name). _(shown when Use Schema Registry is `true`)_ |
| Use Key | `boolean` | No | `false` | Whether to include a message key. |
| Key | `string` | Yes | — | The message key. _(shown when Use Key is `true`)_ |
| JSON Parameters | `boolean` | No | `false` | Whether to provide headers as a JSON string. |
| Headers | `fixedCollection` | No | `{}` | Message headers as key-value pairs. _(shown when JSON Parameters is `false`)_ |
| — Key | `string` | No | — | The header name. |
| — Value | `string` | No | — | The header value. |
| Headers (JSON) | `json` | No | — | Header parameters as a JSON object (flat key-value). _(shown when JSON Parameters is `true`)_ |
| Options | `collection` | No | `{}` | Producer options. |
| — Acks | `options` | No | `all` | Acknowledgement level. "all" waits for all in-sync replicas (safest, required with idempotent producer); "leader" only waits for the partition leader; "none" does not wait. |
| | | | | Options: `all` (durable, wait for ISR), `leader` (wait for leader only), `none` (fire-and-forget) |
| — Compression | `boolean` | No | `false` | Whether to use GZIP compression for messages. |
| — Timeout | `number` | No | `30000` | Time to await a response from the broker in milliseconds (1000-300000). |

This node has no Max Concurrency parameter: sends are grouped by topic and dispatched as batches rather than item by item.

## Output Data

One output item per input item — the node never fans out. The send result is **merged onto the input item JSON**: the fields already on the item pass through, and the node adds two of its own. Binary data is forwarded unchanged.

- `_kafkaTopic` — the topic this item was published to.
- `_kafkaResponse` — an array describing the broker's acknowledgement of the batch this item belonged to. Each entry covers one topic-partition and carries `topicName`, `partition`, `errorCode`, `baseOffset`, `logAppendTime` and `logStartOffset`. When the broker returns no per-partition detail, the array is `[{ "success": true }]` instead.

```json
{
  "_kafkaTopic": "user-events",
  "_kafkaResponse": [
    {
      "topicName": "user-events",
      "partition": 0,
      "errorCode": 0,
      "baseOffset": "1042",
      "logAppendTime": "-1",
      "logStartOffset": "0"
    }
  ]
}
```

Because items are sent in batches, every item in the same batch receives the same `_kafkaResponse` array — it describes the batch, not that one message. Reference it downstream by expression, e.g. `{{ $json._kafkaTopic }}`.

## Usage Examples

- Send JSON data to a Kafka topic
- Publish binary file content to Kafka
- Produce keyed messages with GZIP compression
- Send messages with custom headers
- Encode messages with Confluent Schema Registry

## Example Configuration

Publish each incoming item as a JSON message:

```json
{
  "type": "kafka",
  "parameters": {
    "topic": "user-events",
    "sendInputData": true,
    "options": {
      "acks": "all",
      "timeout": 30000
    }
  }
}
```

Send a fixed message with a partition key and GZIP compression:

```json
{
  "type": "kafka",
  "parameters": {
    "topic": "notifications",
    "sendInputData": false,
    "message": "System maintenance scheduled",
    "useKey": true,
    "key": "system-alert",
    "options": {
      "acks": "leader",
      "compression": true,
      "timeout": 15000
    }
  }
}
```

Publish a file from an upstream node as the message value:

```json
{
  "type": "kafka",
  "parameters": {
    "topic": "file-uploads",
    "sendInputData": false,
    "useBinaryData": true,
    "binaryPropertyName": "data",
    "options": {
      "compression": true,
      "timeout": 45000
    }
  }
}
```

Encode messages with the Confluent Schema Registry and attach headers:

```json
{
  "type": "kafka",
  "parameters": {
    "topic": "customer-data",
    "sendInputData": true,
    "useSchemaRegistry": true,
    "schemaRegistryUrl": "https://schema-registry.example.com:8081",
    "eventName": "customer.profile",
    "jsonParameters": true,
    "headerParametersJson": "{\"service\": \"api-gateway\", \"version\": \"1.2.3\"}",
    "options": {
      "acks": "all",
      "compression": true
    }
  }
}
```

Attach headers as key-value pairs instead of JSON:

```json
{
  "type": "kafka",
  "parameters": {
    "topic": "audit-logs",
    "sendInputData": true,
    "jsonParameters": false,
    "headersUi": {
      "headerValues": [
        { "key": "event-type", "value": "user-login" },
        { "key": "correlation-id", "value": "req-12345" }
      ]
    },
    "options": {
      "timeout": 20000
    }
  }
}
```

### Error Handling

| Mode | Behavior |
|------|----------|
| **stop** | Halts workflow on first error |
| **continue** | Skips failed items, passes successful ones through |
| **errorPort** | Routes failed items to Error output port |

## Tips

Produce messages to Kafka topics with optional binary payloads, schema registry encoding, custom headers, and GZIP compression.