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 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.

{
  "_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:

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

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

{
  "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:

{
  "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:

{
  "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:

{
  "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.