Kafka

📨

Kafka

Produce and consume messages with Apache Kafka

The Kafka node integrates with Apache Kafka clusters for event-driven workflows. You can produce messages to topics, consume messages, and manage topics (create, delete, list, describe). This is useful for event-driven automation, data pipeline orchestration, and inter-system communication.

Before You Use This Node

What you may need	Where to set it up	Why it matters
Kafka cluster connection	Connections	Create the Kafka connection once, then reuse it for publish, consume, and topic actions.
Topic names or standard message values	Global Variables	Keep shared event names and IDs consistent across workflows.
Certificates, if your Kafka setup uses them	Certificates	Store certificate material centrally instead of repeating it in steps.

Actions

Action	Description
kafka:sendMessage	Publish a message to a Kafka topic
kafka:consumeMessages	Consume messages from a Kafka topic
kafka:createTopic	Create a new Kafka topic
kafka:deleteTopic	Delete a Kafka topic
kafka:listTopics	List all topics in the cluster
kafka:describeTopic	Get metadata about a topic (partitions, replicas, config)

---

kafka:sendMessage

Publishes one or more messages to a Kafka topic.

Parameters

Parameter	Type	Required	Description
topic	string	Yes	Target topic name
key	string	No	Message key for partitioning
value	string	Yes	Message value (typically JSON)
headers	object	No	Key-value pairs attached as Kafka message headers
partition	integer	No	Specific partition to send to. If omitted, the default partitioner is used

kafka:consumeMessages

Consumes messages from one or more partitions of a topic. This is a polling operation that reads currently available messages up to the configured limit.

Parameters

Parameter	Type	Required	Description
topic	string	Yes	Topic to consume from
groupId	string	No	Consumer group ID. If omitted, a temporary group is used
maxMessages	integer	No	Maximum number of messages to consume. Defaults to 10
timeout	integer	No	Poll timeout in milliseconds. Defaults to 5000
fromBeginning	boolean	No	Start from the earliest offset. Defaults to false (latest)

Consumer Groups and Offset Management

• When groupId is provided, Kafka tracks offsets for that group. The next consume call picks up where the last one left off, providing at-least-once delivery.
• When groupId is omitted, a temporary group is created. Messages are read but offsets are not committed, so the same messages may be read again on subsequent calls.
• Setting fromBeginning: true resets the consumer to the earliest offset. This is useful for reprocessing all messages in a topic.
• Multiple jobs using the same groupId share the workload across partitions. Each partition is assigned to only one consumer in the group at a time.

kafka:createTopic

Creates a new topic in the Kafka cluster.

Parameters

Parameter	Type	Required	Description
topic	string	Yes	Topic name
partitions	integer	No	Number of partitions. Defaults to 1
replicationFactor	integer	No	Replication factor. Defaults to 1
config	object	No	Topic-level configuration overrides (e.g., retention.ms, cleanup.policy)

kafka:deleteTopic

Deletes a topic from the Kafka cluster. This is irreversible.

Parameters

Parameter	Type	Required	Description
topic	string	Yes	Topic name to delete

kafka:listTopics

Lists all topics in the cluster.

Parameters

No additional parameters required.

kafka:describeTopic

Returns metadata about a topic including partition count, replica assignments, and configuration.

Parameters

Parameter	Type	Required	Description
topic	string	Yes	Topic name

Connection

Requires a Kafka connection configured in the Global Configurator:

Field	Description
bootstrapServers	Comma-separated list of Kafka broker addresses (e.g., broker1:9092,broker2:9092)
securityProtocol	PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL
saslMechanism	PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512 (when using SASL)
saslUsername	SASL username
saslPassword	SASL password
sslTruststoreLocation	Path to SSL truststore (when using SSL)
sslTruststorePassword	Truststore password

Configure under Connections > Kafka in the Global Configurator.

Delivery Guarantees

• sendMessage: At-least-once delivery. The producer waits for broker acknowledgement before reporting success.
• consumeMessages: At-least-once when using consumer groups. The offset is committed after messages are read, so a failure mid-processing may result in redelivery.
• Exactly-once semantics: Kafka does not provide exactly-once delivery out of the box in this integration. To achieve effectively-once processing, implement idempotent logic in your workflow (for example, check whether a record has already been processed before acting on it).

Best Practices

• Use meaningful consumer group names (e.g., order-processor, invoice-sync) rather than generic names like group1.
• Set an appropriate maxMessages value to avoid processing too many messages in a single step. Start small and increase as needed.
• Use message keys for ordering guarantees. Messages with the same key are routed to the same partition, preserving their order.
• Monitor topic lag (the difference between the latest offset and the consumer group's committed offset) to detect processing delays.
• Handle failed message processing by logging the failure and continuing rather than stopping the entire workflow. You can send failed messages to a dead-letter topic for later review.

Common Problems

Symptom	Likely Cause	Fix
Connection refused	Incorrect bootstrapServers or wrong security protocol	Verify broker addresses and that securityProtocol matches the cluster configuration
Authentication failed	Invalid SASL credentials	Check saslUsername, saslPassword, and saslMechanism in the connection settings
No messages consumed	Wrong topic name, missing groupId, or fromBeginning not set	Confirm the topic exists with kafka:listTopics, and set fromBeginning: true if you need to read from the start
Messages consumed but not processed	The step after consume is failing	Check subsequent step logs for errors. The consume itself succeeded
Topic creation failed	Insufficient cluster permissions	Verify the connection credentials have topic-admin privileges on the cluster

Output

sendMessage

consumeMessages

listTopics

describeTopic