Kafka

A FlowFuse-certified package that lets you connect to Apache Kafka, send and receive messages, run consumer groups, commit or roll back messages by hand, manage topics, and read offsets from your flows.

Get the Certified Node in FlowFuse

The Kafka package is part of the FlowFuse Hub Certified Nodes catalogue, which is part of the FlowFuse Hub offering. Please contact our sales team at Contact us to learn more or to request access.

Installation steps

1. Open your instance in the FlowFuse editor.
2. Click the menu icon (☰) in the top-right corner.
3. Select Manage palette.
4. Go to the Install tab.
5. Switch to the FlowFuse Hub Certified Nodes category.
6. Search for @flowfuse-certified-nodes/kafka.
7. Click Install.

The Kafka nodes then appear in your palette, ready to drag onto the canvas.

Note: Newly installed nodes are picked up automatically, no restart needed. Restart is only required when you update a node that's already installed: restart any remote instance or hosted instance running the previous version.

Features

• Send messages to any topic, and set the key, partition, and compression
• Read from several topics at once, each with its own starting offset and partition
• Match topics by pattern, so new topics are picked up without a redeploy
• Consumer groups, so several consumers can share the work
• Commit and roll back by hand, instead of letting Kafka mark messages as done straight away
• Convert / to . in topic names, so MQTT-style topics work with Kafka
• Manage topics from a flow: create, delete, and add partitions
• Read offsets, so you can measure how far behind a consumer is
• TLS, including self-signed certificates, and SASL PLAIN username and password authentication

Use case

Kafka stores messages instead of just passing them on. MQTT sends a message and forgets it. Kafka keeps it for a set period, so a system that was offline can catch up later, and a new consumer can read messages that arrived before it started.

That makes it a good fit for these jobs:

• Send production data to IT systems. Write each reading to a topic with Kafka Producer. An ERP, MES, or data lake reads that topic in its own time. It never connects to your Node-RED instance directly.
• Feed several systems from one topic. Point more than one Kafka Consumer at the same topic. Each one gets every message, and each one keeps its own place in the topic. A dashboard, an alert flow, and a historian can all read the same data without getting in each other's way.
• Ride out an outage. If a system downstream goes down, messages build up in the topic instead of being lost. You can also start a consumer at an older offset to read a shift's data again after fixing a bug.
• Handle more messages. Use Kafka Consumer Group when one consumer cannot keep up. Members of a group split the work rather than each doing all of it.
• Make sure nothing is dropped. Turn off auto commit and use Kafka Commit and Kafka Rollback. A message is only marked as done once your flow has finished with it, so a failure part-way through does not lose data.

Nodes in this package

Where a node has two outputs, the first fires on success and the second on failure. Wire the second one up so errors don't disappear.

Every node has a Name field, which only changes the label shown on the canvas.

Set up the connection (Kafka Broker)

Create one Kafka Broker config node for each Kafka cluster, then point your other nodes at it. You only enter the connection details once, and the nodes that use it share a single connection to the cluster.

Five nodes have a Broker field: Kafka Producer, Kafka Consumer, Kafka Consumer Group, Kafka Admin, and Kafka Offset. Kafka Commit and Kafka Rollback do not, because they work out which connection to use from the message they receive.

1. Drag one of those five nodes onto the canvas and open it
2. Click the + button next to Broker
3. Give the connection a Name, such as Production Kafka
4. Fill in the Connection tab, then the Options tab if you need to
5. Click Add, then Done

Once the connection exists, the other nodes pick it from their own Broker dropdown. Use the pencil button beside that dropdown to edit it later.

Connection tab

Authentication here is SASL PLAIN only, so the username and password travel in the clear unless you also turn on Use TLS. Tick both together on anything but a local test cluster.

You only need one or two entries in Hosts. Kafka uses them to find the cluster, then learns about the remaining brokers on its own, so there is no need to list every broker you run.

Keeping hosts out of your flows

Typing hostnames straight into the table works, but it ties the flow to one cluster. If the same flow runs against test and production, keep the hosts in an instance environment variable instead.

The Hosts process.env name field takes the name of the variable, not its value:

1. Go to Instance Settings → Environment and add a variable, for example KAFKA_HOSTS. Its value is the host list as JSON:
   Copy

   [{"host":"kafka-1.example.com","port":9092}]
   

2. Restart the instance so Node-RED picks up the variable.
3. In the Kafka Broker config node, put KAFKA_HOSTS in the Hosts process.env name field.

Each environment can now point at its own cluster, and the flow stays the same. See Using environment variables in Node-RED for more on environment variables.

Options tab

These control the connection itself rather than anything about your messages. The defaults are sensible, so most people never touch this tab.

On a slow or busy network, raising Connect Timeout and Request Timeout is usually the first thing to try.

The examples on this page use 127.0.0.1 and 9092 as stand-ins for a local Kafka broker. Replace them with your own details.

Send messages (Kafka Producer)

Kafka Producer sends msg.payload to a topic. It has one input and no output.

Key is a typed input, so you can enter a fixed string or point it at a message property.

Require Ack is the setting worth thinking about. 0 is fastest and can lose messages silently. 1 is the usual compromise. -1 is the safest and the slowest, and only makes sense on a topic with more than one replica.

Partitioner options

The partitioner decides which partition a message goes to. It only comes into play when the message itself doesn't say: set Partition on the node, or msg.partition on the message, and the partitioner is bypassed.

Pick keyed when order matters within a group of messages — readings from one machine, events for one order number — and set Key to whatever identifies that group. Kafka only guarantees order inside a partition, so this is the setting that keeps a machine's readings sequential.

Pick cyclic when you only care about spreading work evenly and order doesn't matter.

default is worth a second look if you have a multi-partition topic and find everything piling into partition 0.

Compression options

The Compression dropdown offers:

Where a codec has best speed and best compression variants, you are trading CPU time against message size. The plain option sits between the two.

Whatever you pick, whatever reads the topic has to understand it. If your consumers are also Kafka Consumer nodes, set the same codec on their Decompression field. If anything else reads the topic, stick to gzip or Snappy, which every Kafka client handles. The other codecs here are not part of the standard Kafka set, so a Java or Python consumer may not be able to read them.

Setting values from the message

Four of these fields can come from the incoming message instead of the node:

• msg.topic — the topic to write to
• msg.key — the message key
• msg.partition — the partition to write to
• msg.attributes — the attribute flag sent with the message. There is no field for this in the dialog, so the message is the only way to set it

If the message does not carry one of these, the node's own value is used instead. The node settings are defaults, in other words, and any message can override them.

This is what lets a single producer node serve many topics. Leave Topic blank on the node, set msg.topic in a change node beforehand, and the same producer handles all of them.

Say you have a Temperature topic with three partitions, carrying downtown on partition 0, suburban on 1, and industrial on 2, with the region name as the key. You can build that with three producer nodes, each configured for its own region, or with one node and a change node that sets msg.key and msg.partition.

Receive messages (Kafka Consumer)

Kafka Consumer has no input. It sends out one msg for every Kafka message that arrives.

Two fields sit above the tabs, alongside Name:

The remaining settings are split across four tabs: Topics, Options, Fetch, and Encoding.

Topics tab

Add a row to the Topic table for each topic you want, using +Add. Each row has three columns:

Tick wildcard (regex) to treat the names as patterns instead of exact topics. Matching topics are picked up as they appear and dropped when they go away. Kafka is checked once a minute, so a new topic can take up to a minute to show up.

This is the setting to reach for when a topic is created per machine or per line. A pattern like line.*\.temperature keeps working as lines are added, with no redeploy.

Options tab

Fetch tab

How much data the consumer pulls at a time. Leave these alone unless you are chasing throughput or latency.

Raising Fetch Min Bytes and Fetch Max Wait means fewer, fuller round trips, which suits high message rates. Lowering them gets messages to your flow sooner, which suits alerting.

Encoding tab

Pick earliest on From Offset when a new consumer needs the history that is already in the topic. Pick latest when it should only see what happens from now on.

What arrives in the message

Every message the node sends out carries two things:

• msg.payload — the message itself. With the default utf8 encoding this is text, unless you turn on Convert message to JSON. See Payloads are text
• msg._kafka — the details from Kafka: which topic the message came from, the partition and offset it was read at, and the key set by the producer as msg._kafka.key

Keep msg._kafka on the message if you plan to commit or roll back later, because Kafka Commit and Kafka Rollback both need it.

Share the work (Kafka Consumer Group)

Kafka Consumer Group reads messages the same way as Kafka Consumer, but members that use the same Group Id split the work between them. Kafka gives each topic and partition to one member only, and hands them out again when a member joins or leaves.

Use a group when you need to handle more messages. Use separate Kafka Consumer nodes when every consumer needs to see every message.

Above the tabs it has Type and Broker, the same as Kafka Consumer. Type only offers Base, so there is nothing to decide there.

Topics tab

Add a row for each topic. Unlike Kafka Consumer, you only give the Name here — there is no per-topic offset or partition.

Options tab

A group cannot split the work more ways than there are topics and partitions. Any extra members sit idle, which is worth remembering before adding a fourth consumer to a three-partition topic.

There is no Auto Commit setting here, and no Convert message to JSON. Use a json node in the flow if you need to parse the payload.

Commit and roll back (Kafka Commit, Kafka Rollback)

By default a consumer commits automatically, so Kafka treats a message as done as soon as it hands it over. If your flow then fails, the message is gone. Committing by hand instead means a message is only marked as done once your flow has finished with it.

What each node does

Kafka Commit commits a message that was read without auto commit. Put it at the end of the path that worked.

Kafka Rollback rolls the message back and closes the consumer that read it. The message stays uncommitted and Kafka stops handing it out. Put it on the error path, then fix the problem and open the consumer again to carry on.

What they need to work

• Auto commit must be off. Set Auto Commit to False on the Kafka Consumer node. Neither node does anything useful while the consumer is committing on its own.
• The message must still carry msg._kafka. Both nodes use it to find the message and its consumer. A change or function node that rebuilds msg from scratch will remove it.
• They only work with Kafka Consumer. Kafka Consumer Group has no auto commit setting to turn off.

Neither node has a Broker field, or any setting beyond Name — they get the connection from the message. Both have two outputs: the first for success, the second for failure.

Note

Kafka keeps sending messages to a consumer whether or not an earlier commit is still outstanding. A commit therefore marks every message up to that point as done, not just the one that triggered it. If that matters to you, keep the number of messages in flight low.

Administration (Kafka Admin)

Kafka Admin runs one admin command against the cluster. These are the same jobs you would otherwise do with the kafka-topics.sh and kafka-configs.sh scripts on a broker. Running them from a flow means you can create topics as part of a deployment, show cluster state on a dashboard, or check consumer groups on a schedule.

The node has only Name and Broker. Everything else comes from the message. It has two outputs, one for success and one for failure.

How to send a command

Every command uses the same two message properties:

• msg.topic — the command name, spelled as in the tables below. Capital letters matter
• msg.payload — the command's arguments, as JSON. Leave it out for commands that take none

The reply comes back on msg.payload from the first output.

Watch out for one thing: msg.topic holds the command name, not a Kafka topic name. The topic a command acts on goes in msg.payload. This trips up most people the first time.

An inject node is the easiest way to drive it. Set its Topic to the command and its Payload to the JSON, wire it into Kafka Admin, and add a debug node to each output so you see both the reply and any error.

Commands that take no arguments

Send these with msg.topic set and no payload.

Commands that take arguments

Arguments always go in msg.payload.

Examples

Create a topic with three partitions. Set msg.topic to createTopics and msg.payload to:

[{ "topic": "Temperature", "partitions": 3, "replicationFactor": 1 }]

The payload is an array, so one message can create several topics at once. Add configEntries to set topic settings at the same time, such as keeping messages for seven days:

[{
  "topic": "Temperature",
  "partitions": 3,
  "replicationFactor": 1,
  "configEntries": [{ "name": "retention.ms", "value": "604800000" }]
}]

List every topic. Set msg.topic to listTopics and send no payload.

Add partitions to a topic. Set msg.topic to createPartitions. You can only ever add partitions, never remove them, so this cannot be undone. Adding partitions also changes which partition each key goes to, so existing keys stop landing where they used to.

Delete a topic. Set msg.topic to deleteTopics and put the topic names in the payload. This only works if the brokers have delete.topic.enable turned on, which is the default from Kafka 1.0 onwards. If it is off, the request is accepted and then ignored.

Things to watch out for

Creating a topic takes a moment. The reply means the request was accepted, not that every broker knows about the topic yet, so a listTopics sent straight afterwards may not include it. If a flow creates a topic and then writes to it, add a short delay rather than wiring the two together directly.

replicationFactor cannot be higher than your number of brokers. Asking for 3 on a single-broker test cluster will fail, so use 1 locally and set the real value per environment with an environment variable.

You may not need to create topics at all. If the brokers have auto.create.topics.enable turned on, writing to an unknown topic creates it with the default settings. That is handy while testing and usually unwanted in production, because a typo quietly creates a new topic instead of failing.

Admin commands also need the right permissions. On a cluster with ACLs, the credentials in your Kafka Broker config need cluster-level rights, so read-only credentials that work fine for a consumer will not work here.

Offsets (Kafka Offset)

Kafka Offset reads offset details from Kafka. Like Kafka Admin, the node has only Name and Broker, takes everything else from the message, and has two outputs.

It uses two message properties:

• msg.action — which of the four actions to run. You can also put this in msg.topic
• msg.payload — the arguments for that action

Payloads

fetch takes a partition and a limit:

[{ "topic": "test", "partition": 0, "maxNum": 100 }]

fetchEarliestOffset and fetchLatestOffsets just take the topic:

[{ "topic": "test" }]

fetchCommits is the odd one out. Its payload is an object rather than an array, and it needs the group as well as the topics:

{
  "groupid": "aGroup",
  "topics": [{ "topic": "test" }]
}

Note the spelling: groupid, all lower case.

Measuring how far behind a consumer is

Run fetchLatestOffsets for the topic and fetchCommits for the group reading it, then subtract one from the other. The gap is how many messages the group still has to get through. Raise an alert when it grows past a threshold you are comfortable with.

These are also the offsets you put in a consumer's Topic table when you want to read a period of data again.

Payloads are text

Kafka stores messages as bytes. With the default utf8 encoding, a consumer gives you the message as text, and a producer expects text.

The nodes can handle the conversion for you, or you can do it in the flow:

• Sending: turn on Convert message from JSON on Kafka Producer, or put a json node in stringify mode before it.
• Receiving: turn on Convert message to JSON on Kafka Consumer, or put a json node in parse mode after it.

Kafka Consumer Group has no equivalent option, so use a json node there.

For a plain number, a json node is more than you need. A change node that sets msg.payload to the JSONata expression $number(msg.payload) is enough.

For binary payloads, set Encoding to raw on the consumer and skip the json node altogether.