Kafka Integration

The Apache Kafka integration connects ThingsBoard to Apache Kafka message brokers. Use it to consume records that producers — devices, IoT platforms, connectivity providers, or your own applications — publish to Kafka topics, and bring them into ThingsBoard as telemetry and attributes for monitoring and rule-based automation — including publishing commands back to Kafka.

A producer publishes a record to a Kafka topic. The Kafka Integration, acting as a Kafka consumer, subscribes to the topic and polls the broker for new records. The uplink converter decodes the record value into telemetry and attributes, and the ThingsBoard Core Service stores the data, creating the device automatically on first contact. For downlink, the Rule Engine hands a message to a Kafka Rule Node, which publishes it to a separate Kafka topic; external consumers then read it. The Kafka Integration has no dedicated downlink converter.

Before creating the integration, ensure:

• You have access to ThingsBoard Cloud with integration functionality enabled for your tenant.
• You have permissions to create integrations and data converters.
• You have a running Apache Kafka broker — either a local installation or a cloud-managed cluster such as Confluent Cloud. If you do not have a cluster yet, follow the Kafka Setup step below.

The Kafka Integration connects ThingsBoard to an existing Apache Kafka broker as a consumer. The steps below stand up a minimal single-node broker in KRaft mode (no ZooKeeper), which is enough to test the integration. If you use a managed cluster (Confluent Cloud, Amazon MSK, Aiven, Redpanda Cloud, and others), skip to Required Connection Details and obtain the values from your provider instead.

KRaft is Kafka’s built-in consensus protocol; since Kafka 4.0 it fully replaces ZooKeeper, so one process acts as both controller and broker. Kafka 4.x requires Java 17+, and formatting storage is a required, explicit step before the first start. The commands below were verified on Ubuntu with Kafka 4.3.1 — adjust the version and paths for your environment.

1. Install Java and download Kafka:

sudo apt update
sudo apt install -y openjdk-17-jre-headless curl
java -version

cd /tmp
curl -fLO https://downloads.apache.org/kafka/4.3.1/kafka_2.13-4.3.1.tgz
sudo tar -xzf kafka_2.13-4.3.1.tgz -C /opt
sudo ln -s /opt/kafka_2.13-4.3.1 /opt/kafka
sudo chown -R "$USER":"$USER" /opt/kafka_2.13-4.3.1
cd /opt/kafka

2. Initialize KRaft storage (run once per broker):

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format \
  --standalone \
  --cluster-id "$KAFKA_CLUSTER_ID" \
  --config config/server.properties

3. Start the broker:

bin/kafka-server-start.sh config/server.properties

Create the uplink topic the integration will subscribe to (or enable Auto create topics in the integration to have ThingsBoard create it on subscribe). Run the command from the Kafka install directory (e.g. /opt/kafka):

bin/kafka-topics.sh --create \
  --topic my-topic \
  --bootstrap-server localhost:9092 \
  --partitions 1 \
  --replication-factor 1

Verify it exists (optional):

bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

Collect these values for the integration’s Connection step. For the single-node broker above, they are:

If the broker requires authentication — most managed and cloud brokers do — you also need credentials, typically a username/password or an API key/secret, together with the security protocol (e.g. SASL_SSL). These are entered under Advanced settings ⇾ Other properties as SASL/SSL parameters (see the Other Properties row under Connection Settings). For an open or local broker without authentication, no credentials are needed — leave these empty.

The uplink converter receives each consumed Kafka record, decodes the record value, and returns a structured object that ThingsBoard uses to create or update a device and store its telemetry and attributes. For the full decoder function reference — all input parameters and output fields — see Uplink Data Converter.

Kafka uses a generic uplink converter. These converters decode a JSON message received from a Kafka topic and transform it into the ThingsBoard integration data format.

1. Go to Integrations center ⇾ Data converters.
2. Click + Add data converter ⇾ Create new converter.

In the Add data converter dialog:

3. Converter type — leave Uplink (selected by default).
4. Integration type — in the search field, enter Kafka and select Kafka from the list.
5. Name — enter a converter name, for example Kafka Uplink Converter.
6. Main decoding configuration — a code editor with the function signature function decoder(payload, metadata) {. Paste the decoder function shown below. By default the editor opens in TBEL; use the TBEL / JS toggle (upper right) to switch languages.
7. Click Add.

/** Decoder **/

// Decode the Kafka message payload into a JSON object.
var payloadJson = decodeToJson(payload);

// Use the EUI value as the ThingsBoard device name.
var deviceName = payloadJson.EUI;

// Assign the same device type to all devices created by this integration.
var deviceType = 'Monitoring sensor';

// Build the object expected by the ThingsBoard integration.
var result = {
  deviceName: deviceName,
  deviceType: deviceType,

  // No device attributes are extracted from the incoming message.
  attributes: {},

  // Create a timestamped telemetry entry.
  telemetry: {
    // Use the timestamp provided in the Kafka message.
    ts: payloadJson.ts,

    values: {
      // Read the first byte from the hexadecimal data field.
      Temperature: hexToInt(payloadJson.data.substring(0, 2)),

      // Read the second byte from the hexadecimal data field.
      Humidity: hexToInt(payloadJson.data.substring(2, 4)),

      // Read the third byte from the hexadecimal data field.
      Fan: hexToInt(payloadJson.data.substring(4, 6)),

      // Copy message metadata directly to telemetry.
      Port: payloadJson.port,
      Freq: payloadJson.freq,

      // Read the fourth byte from the hexadecimal data field.
      Pressure: hexToInt(payloadJson.data.substring(6, 8)),

      // Copy signal strength and serial number to telemetry.
      rssi: payloadJson.rssi,
      serial: payloadJson.serial
    }
  }
};

// Converts a hexadecimal string into an integer.
function hexToInt(hex) {
  return parseInt(hex, 16);
}

/** Helper function 'decodeToJson' is already built-in **/

return result;

/** Decoder **/

// Decode the Kafka message payload into a JSON object.
var payloadJson = decodeToJson(payload);

// Use the EUI value as the ThingsBoard device name.
var deviceName = payloadJson.EUI;

// Assign the same device type to all devices created by this integration.
var deviceType = 'Monitoring sensor';

// Build the object expected by the ThingsBoard integration.
var result = {
  deviceName: deviceName,
  deviceType: deviceType,

  // No device attributes are extracted from the incoming message.
  attributes: {},

  // Create a timestamped telemetry entry.
  telemetry: {
    // Use the timestamp provided in the Kafka message.
    ts: payloadJson.ts,

    values: {
      // Read the first byte from the hexadecimal data field.
      Temperature: hexToInt(payloadJson.data.substring(0, 2)),

      // Read the second byte from the hexadecimal data field.
      Humidity: hexToInt(payloadJson.data.substring(2, 4)),

      // Read the third byte from the hexadecimal data field.
      Fan: hexToInt(payloadJson.data.substring(4, 6)),

      // Copy message metadata directly to telemetry.
      Port: payloadJson.port,
      Freq: payloadJson.freq,

      // Read the fourth byte from the hexadecimal data field.
      Pressure: hexToInt(payloadJson.data.substring(6, 8)),

      // Copy signal strength and serial number to telemetry.
      rssi: payloadJson.rssi,
      serial: payloadJson.serial
    }
  }
};

/** Helper functions **/

// Converts the incoming byte array into a UTF-8-compatible string.
function decodeToString(payload) {
  return String.fromCharCode.apply(String, payload);
}

// Converts the incoming byte array into a string and parses it as a JSON object.
function decodeToJson(payload) {
  var str = decodeToString(payload);
  return JSON.parse(str);
}

// Converts a hexadecimal string into an integer. Matches the TBEL tab —
// both read a single byte (2 hex characters) at a time, so no byte-order
// reversal is needed. If you extend this to multi-byte fields, decide on
// endianness explicitly and keep both tabs consistent.
function hexToInt(hex) {
  return parseInt(hex, 16);
}

return result;

What the Converter Receives

ThingsBoard passes two variables to the decoder function:

Example: JSON Payload Decoded

Apache Kafka does not impose an application-level schema on a record’s value. The value is stored as bytes and may contain JSON, Avro, Protobuf, plain text, or any other format selected by the producer.

This example models a typical IoT data flow in which a LoRaWAN Network Server, or an intermediate service, publishes device uplink events to Kafka. Therefore, the sample message contains LoRaWAN-related fields such as a device EUI, frame port, frequency, and RSSI rather than Kafka-specific fields.

Note that the field names and message structure used in this example are not defined by Kafka or standardized across all LoRaWAN Network Servers. Each producer may use a different schema, naming convention, nesting structure, or serialization format. Update the uplink converter and decoder to match the exact message format produced by your system.

In this example, the device 43T1YH-REE publishes a JSON record to Kafka. The sensor readings are encoded as a 4-byte hexadecimal string in the data field, and the incoming message is expected to contain the following fields:

{
  "EUI": "43T1YH-REE",
  "ts": 1638876127000,
  "data": "3d1f0059",
  "port": 10,
  "freq": 24300,
  "rssi": -130,
  "serial": "230165HRT"
}

The data field must contain at least eight hexadecimal characters. Each pair of characters represents one telemetry value:

The converter returns:

{
  "deviceName": "43T1YH-REE",
  "deviceType": "Monitoring sensor",
  "attributes": {},
  "telemetry": {
    "ts": 1638876127000,
    "values": {
      "Temperature": 61,
      "Humidity": 31,
      "Fan": 0,
      "Port": 10,
      "Freq": 24300,
      "Pressure": 89,
      "rssi": -130,
      "serial": "230165HRT"
    }
  }
}

In short:

• deviceName: the value of the EUI field;
• deviceType: the static Monitoring sensor device type;
• attributes: an empty object because the message does not contain attribute data;
• telemetry: timestamped telemetry values extracted from the message.

1. Go to Integrations center ⇾ Integrations and click + Add integration.
2. Basic settings:
   • Set Integration type to Kafka.
   • Enter a Name for the integration, or keep the default Kafka integration.
   • Leave Enable integration and Allow create devices or assets on so a device is created automatically on first message.
   • Click Next.
3. Uplink data converter:
   • Click Select existing and choose the Kafka Uplink Converter created above, or click Create new to define the decoder inline.
   • Click Next.
4. Connection:
   • Fill in Group ID, Client ID, Bootstrap servers, Poll interval, and Topics — see Required Connection Details for the values to use.
   • Toggle Auto create topics if the topic does not exist yet.
   • For a secured broker, expand Advanced settings and add SASL/SSL properties under Other properties.
   Read more about each parameter in connection settings.
5. Click Add.

Set Integration type to Kafka, enter a name, and click Next Click Select existing, choose Kafka Uplink Converter, and click Next Fill in Group ID, Client ID, Bootstrap servers, Poll interval, and Topics, then click Add

Publish a test record to the topic the integration subscribes to, then optionally confirm it landed on the topic. Run the commands from the Kafka install directory (e.g. /opt/kafka).

Use kafka-console-producer.sh to publish a test message. Replace localhost:9092 with your broker address and my-topic with your configured topic if they differ:

echo '{"EUI":"43T1YH-REE","ts":1638876127000,"data":"3d1f0059","port":10,"freq":24300,"rssi":-130,"serial":"230165HRT"}' | \
  bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic > /dev/null

Verify the Record on the Topic (Optional)

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning

You should see the record echoed back:

{"EUI":"43T1YH-REE","ts":1638876127000,"data":"3d1f0059","port":10,"freq":24300,"rssi":-130,"serial":"230165HRT"}

Go to Integrations center ⇾ Integrations, open Kafka integration, and click the Events tab. With Debug mode on, each consumed record appears as an event with Type: Uplink and Status: OK. Click the ⋮ icon in the Message column to inspect the raw record the integration received from Kafka.

Integration → Events — confirm the Uplink event with Status OK Click the Message column to inspect the raw record

Go to Integrations center ⇾ Data converters, open Kafka Uplink Converter, and click the Events tab. Click In to see the raw Kafka record the decoder received (EUI, ts, data, port, freq, rssi, serial), and Out to see the decoded deviceName, deviceType, attributes, and telemetry it produced.

Data converter → Events — one row per consumed record Click In to see the raw record the decoder received Click Out to see the decoded deviceName, deviceType, and telemetry values

Go to Entities ⇾ Devices — device 43T1YH-REE (Device profile: Monitoring sensor) is auto-created on the first record. Open it and check the Latest telemetry tab — you should see all 8 decoded keys: Temperature, Humidity, Fan, Port, Freq, Pressure, rssi, and serial.

The Kafka Integration has no dedicated downlink converter. To publish messages from ThingsBoard back to Kafka, use the Kafka Rule Node in the Rule Engine. The node takes the message payload, publishes it as the Kafka record value, and on broker acknowledgment adds offset, partition, and topic to the message metadata.

The steps below test downlink entirely on the local setup — self-hosted ThingsBoard PE and the Kafka broker on the same machine. The idea: add a Kafka rule node that publishes to a separate downlink topic, then trigger it by updating a shared attribute on device 43T1YH-REE and confirm the value with a console consumer.

1. Create a Separate Downlink Topic

Run from the Kafka install directory (e.g. /opt/kafka):

bin/kafka-topics.sh --create \
  --topic my-topic-downlink \
  --bootstrap-server localhost:9092 \
  --partitions 1 \
  --replication-factor 1

2. Start a Consumer on the Downlink Topic

In a second terminal, leave a consumer running so you can watch messages arrive:

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic-downlink \
  --from-beginning

Add and Configure the Kafka Rule Node:

1. Go to Rule chains and open the Root Rule Chain (the chain that processes your device’s messages).

2. In the node palette on the left, search for kafka — it appears under the External category — and drag it onto the canvas. In the node dialog, set:

   Field	Value for this setup
   Name	Kafka Downlink
   Topic pattern	my-topic-downlink
   Key pattern	(leave empty)
   Bootstrap servers	localhost:9092
   Automatically retry times if fails	0 (or default)
   Produces batch size in bytes	16384 (default)
   Time to buffer locally (ms)	default
   Client buffer max size in bytes	default
   Number of acknowledgments	default
   Other properties	(leave empty — local PLAINTEXT broker needs no auth)

3. Click Add. The Topic pattern supports templates (e.g. ${deviceType}-downlink), but a static name is fine for testing.

Wire the Node into the Rule Chain:

The Kafka node publishes whatever message reaches it, so connect it to a message flow. For a quick end-to-end test, reuse the uplink you already have working:

1. On the canvas, drag a connection from the Message Type Switch node to the new Kafka node.
2. In the relation dialog, select the Attributes Updated label and confirm.
3. Click Apply changes (top-right) to save the rule chain.

Search for kafka in the node palette — it appears under External Add rule node: kafka — Name, Topic pattern, and Bootstrap servers filled in Message Type Switch connected to Kafka Downlink via Attributes Updated

Now, whenever a shared attribute is updated on a device, its value is published to my-topic-downlink.

Update a shared attribute on device 43T1YH-REE:

1. Go to Entities ⇾ Devices and open 43T1YH-REE.
2. Open the Attributes tab and set the scope selector to Shared attributes.
3. Click +, enter key powerState, value on (type String), and click Add.

The attribute update flows through the rule chain, and the Kafka node publishes it to the downlink topic.

Switch to the consumer terminal from Start a Consumer on the Downlink Topic — the record should appear on my-topic-downlink:

{"powerState":"on"}

Open the Root Rule Chain, click the Kafka Downlink node, and open its Events tab. Each trigger produces two events — an IN event (Relation Type: Attributes Updated) for the message entering the node, and an OUT event (Relation Type: Success) for the published record. Click Data on the OUT event to see the published payload (e.g. {"powerState":"on"}), and Metadata to see the publish result — offset, partition, topic, the attribute scope (SHARED_SCOPE), and the user who made the change.

The Kafka Downlink node Events tab — one IN and one OUT event per trigger Click Data on the OUT event to see the published payload Click Metadata to see the publish result — offset, partition, topic, and scope

This section covers the most common problems encountered when setting up and running the Kafka integration. Each entry describes the symptom, the most likely cause, and the steps to resolve it.

No Records Consumed

Record Consumed but Device Not Created

Downlink Not Published

How to Read Debug Events

1. Go to Integrations center ⇾ Integrations, open Kafka integration, and click the Events tab.
2. Click an event row to inspect:

• In — the raw record value received from Kafka before processing.
• Out — what the converter returned: device name, attributes, and telemetry values passed to ThingsBoard.
• Error — error text and stack trace, if processing failed. Enable Debug mode on the integration to capture all raw input/output events. Starting from ThingsBoard 3.9, full debug events are stored only during the first hour — afterwards, only error events are retained. Disable debug mode once the issue is identified.

• Integrations Overview — how ThingsBoard connects to external platforms and how uplink/downlink flow works
• Uplink Data Converter — full decoder function reference: input parameters, output fields, and scripting patterns
• Kafka Rule Node — publish messages from the Rule Engine back to a Kafka topic (downlink)
• Remote Integration — run the integration outside the ThingsBoard server to reach a broker on a private network
• TBEL scripting reference — built-in functions and operators for writing converter scripts
• Rule Engine — how the rule chain routes messages to the Kafka Rule Node