Kafka integration

TBMQ Kafka Integration enables communication with Apache Kafka, allowing TBMQ to publish messages to external Kafka clusters. This is useful for the following scenarios:

• Streaming IoT Data — forwarding device telemetry, logs, or events to Kafka for processing and storage.
• Event-Driven Architectures — publishing messages to Kafka topics for real-time analytics and monitoring.
• Decoupled System Communication — using Kafka as a buffer between TBMQ and downstream applications.

The two technologies complement rather than replace each other — see MQTT vs Kafka for how they compare and why they are so often paired.

1. Device (client) publishes an MQTT message to a topic matching the integration’s topic filters.
2. TBMQ broker receives the message and forwards it to the TBMQ Integration Executor.
3. Integration Executor processes the message and sends it to a configured Kafka topic.
4. Kafka consumers process the message in downstream systems.

• A running TBMQ instance, version 2.1 or later, with the TBMQ Integration Executor service running. Without a reachable executor, saving an integration fails with a validation timeout.
• An external Kafka cluster ready to receive messages (e.g., Confluent Cloud).
• A client capable of publishing MQTT messages (e.g., TBMQ WebSocket Client).

1. Go to the Integrations page and click the “+” button.
2. Select Kafka as the integration type and click Next.
3. On the Filters and events step, keep the default Topic filters entry tbmq/#, leave Lifecycle events empty, and click Next.
4. In the Configuration step, enter the Bootstrap servers (Kafka broker addresses). See below for common and Confluent Cloud configurations. Leave Topic at its default tbmq.messages — that is the topic this tutorial reads from at the end.
5. Click Add to save the integration.

Navigate to Integrations page and click the plus icon Select Kafka as the integration type and click Next Click Next to subscribe to default topic tbmq/#

• Common / Docker
• Confluent Cloud

Specify the bootstrap server address of your own Kafka cluster (e.g., localhost:9092 for a broker you run locally — not TBMQ’s internal Kafka, which uses the same address in the default deployments). The screenshot below shows the basic configuration for establishing a connection between TBMQ and a Kafka broker.

Topic filters define MQTT-based subscriptions that trigger the integration. When TBMQ receives a message matching a configured topic filter, the integration processes it and forwards the data to the Kafka cluster.

Topic filters are optional if you select at least one lifecycle event type instead — an integration only has to have one of the two.

For example, with the topic filter tbmq/devices/+/status, any of the following messages will trigger the integration:

tbmq/devices/device-01/status
tbmq/devices/gateway-01/status

With Send only message payload enabled, the record value is the payload alone — as UTF-8 text if it is valid UTF-8, Base64-encoded otherwise.

The Key and the Kafka headers values can contain ${…} placeholders, filled in for each record from the message, or from the event payload for a client lifecycle event. They work with Send only message payload on, too: the record value stays the raw payload, and the key and headers still carry the message’s metadata.

For example, with Key ${clientId}, the headers mqtt-topic = ${topicName} and username = ${username}, and Send only message payload on, each record carries:

• the raw payload as its value;
• the publishing client’s ID as its key, so one client’s records stay on one partition and in order;
• mqtt-topic and username headers.

A key or header value without placeholders is sent as is, the same for every record.

Everything after the first dot is the key, so ${props.app.version} reads the user property app.version. Placeholder names are case-sensitive.

If any placeholder in a value has no value for a record, the whole value counts as empty: the record is sent without a key, or that header is left out. For example, a key dev-${username} from a client that sent no username gives no key at all, not dev-. Expect this for ${username} and ${clientCertCn} when the client did not provide them, for ${props.<name>} when the message lacks that property, and for any placeholder used where it is not available — such as ${topicName} on a lifecycle event.

With Forward user properties as headers on, every MQTT 5 user property of a message is added as a Kafka header with the same name and value:

• The properties keep the order the client sent them in, and a name the client repeated becomes several headers with that name.
• They come after the configured Kafka headers.
• A name used by a configured header is reserved. A user property with that name is not forwarded, even when the configured header has no value for that record and is left out. Otherwise a client could forge a configured header such as username, because consumers usually read the last header of a given name.

Lifecycle events carry no user properties, so the option affects messages only.

The key and header values are checked when the integration is saved. An unknown, empty (${}), or unclosed placeholder, or ${props} / ${metadata} without a key, rejects the save with an error that names the field:

Key: unknown placeholder '${topic}'. Allowed: ${clientCertCn}, ${clientId}, ${eventType}, ${ipAddress}, ${qos}, ${retain}, ${sessionId}, ${tbmqIeNode}, ${tbmqNode}, ${topicName}, ${ts}, ${username}, ${props.KEY}, ${metadata.KEY}

There is no way to escape ${, so a key or header value cannot contain it as literal text. Enabling or disabling an integration from the Integrations table saves it, too, so the same check applies there.

Besides the messages matched by topic filters, this integration can deliver client lifecycle events — a client connecting, disconnecting, changing its subscriptions, or failing authentication or authorization.

Pick the event types on the Filters and events step while creating the integration, or add them later:

1. Open the integration on the Integrations page and click the Toggle edit mode button (pencil icon).
2. Click the Lifecycle events field to list the available event types, then select the ones you need: Client connected, Client disconnected, Client subscribed, Client unsubscribed, Client authentication failed, Client authorization failed, Client connection failed. Each one is added as a chip; remove it with its x.
3. Click Apply changes.

Events are published to the same Topic as messages, using the configured Key, Kafka headers, and producer settings. Send only message payload applies to message payloads only — an event is always published as its full JSON body.

Placeholders in the key and headers are filled in from the event payload, so a message-only placeholder such as ${topicName} leaves an event without a key, or drops that header — see Placeholders. Forward user properties as headers does not apply to events.

These are the integration’s own events, recorded inside TBMQ for debugging and troubleshooting. They are not the client lifecycle events described above: those are about MQTT clients and are published to your Kafka topic, while these describe the integration itself and never leave TBMQ.

• Lifecycle Events — logs events such as Started, Created, Updated, Stopped.
• Statistics — insights into integration performance, including processed message counts and error rates.
• Errors — captures failures related to authentication, timeouts, payload formatting, or connectivity issues.

Lifecycle events for the Kafka integration Statistics events showing processed message counts Error events capturing integration failures

1. Navigate to the WebSocket Client page.
2. Select WebSocket Default Connection (or any working connection) and click Connect. Verify the connection status shows Connected.
3. Set the Topic field to tbmq/kafka-integration to match the integration’s topic filter tbmq/#.
4. Leave the pre-filled Payload {"temperature": 25} as is — it is what lands in the Kafka topic below.
5. Click the Send icon to publish the message.

Select a working connection and click Connect Set topic to tbmq/kafka-integration and click Send In Confluent Cloud open Topics and click on the topic tbmq.messages The message from TBMQ Kafka integration is available in the messages table

If successful, the message should be available in your Kafka service under the topic tbmq.messages:

{
    "payload": "eyJ0ZW1wZXJhdHVyZSI6MjV9",
    "topicName": "tbmq/kafka-integration",
    "clientId": "tbmq_df52bNUQ",
    "eventType": "PUBLISH_MSG",
    "qos": 1,
    "retain": false,
    "tbmqIeNode": "tbmq_ie_node",
    "tbmqNode": "tbmq_node",
    "ts": 1742554969254,
    "username": "tbmq_websockets_username",
    "props": {},
    "metadata": {
        "integrationName": "Kafka integration"
    }
}

Message field descriptions: