Kafka Integration

Kafka Integration connects ThingsBoard to an Apache Kafka cluster. ThingsBoard joins a consumer group, subscribes to the topics you configure, and polls them for new records. An uplink data converter transforms each record value into device or asset data, including telemetry and attributes.

Use Kafka Integration when:

• Devices, network servers, or your applications already publish data to Kafka topics.
• You want that data in ThingsBoard dashboards and the Rule Engine without a separate bridge service.

The integration is uplink-only. To publish data from ThingsBoard to Kafka, use the kafka rule node (see Send Data to Kafka).

1. Receive data. A producer publishes a record to a Kafka topic. The integration polls the topic as a member of its consumer group and receives the record.
2. Decode the payload. The uplink converter decodes the record value and returns the device name, telemetry, and attributes.
3. Process the message. ThingsBoard processes the converted data through the Rule Engine. If Allow create devices or assets is enabled, ThingsBoard creates the device on its first record.
4. Publish data to Kafka (optional). The Rule Engine passes a message to a kafka rule node, which publishes it to a Kafka topic for external consumers.

• Access to ThingsBoard Cloud with integrations enabled for your tenant.
• Permission to create integrations and data converters, and to edit rule chains to publish data to Kafka.
• An Apache Kafka broker that ThingsBoard can reach: a self-hosted broker or a managed cluster such as Confluent Cloud or Amazon MSK. To start a local test broker, see Set Up Kafka.

Start a broker, create the uplink topic, and collect the connection details. With a managed cluster, skip to Collect Connection Details and take the values from your provider.

Kafka 4.x runs in KRaft mode: one process acts as both controller and broker, without ZooKeeper. The broker needs Java 17 or later. The commands below were verified on Ubuntu with Kafka 4.3.1. For another release, replace the version in the download URL and paths; the Kafka downloads page lists the current ones.

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. Format the storage. Run this once per broker, before the first start:

   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

Run the commands in the next sections from /opt/kafka, in a second terminal.

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

To skip this step, turn on Auto create topics in the integration.

ThingsBoard must reach every broker of the cluster. localhost:9092 works only when ThingsBoard runs on the broker host. Otherwise:

• Set Bootstrap servers to the broker’s reachable host name or IP.
• Set the broker’s advertised.listeners to the same address. The integration connects to the bootstrap address, then fetches records from the advertised address. If the two differ, the connection succeeds, but no records arrive.
• On ThingsBoard Cloud, use a public address, or run the integration next to the broker as a Remote Integration.

Managed clusters usually require SASL over TLS. Add the Kafka client properties under Advanced settings ⇾ Other properties. For SASL/PLAIN with an API key, as on Confluent Cloud:

Other mechanisms, such as SCRAM or mutual TLS, use other keys. Take them from your provider’s client configuration.

Create the uplink converter, assign it to a new Kafka integration, and publish a test record.

The uplink converter receives each record value as a byte array, resolves the target device, and returns its telemetry and attributes. Kafka Integration uses the Generic converter: you write the decoder in TBEL or JavaScript. For converter types and the full decoder reference, see Uplink data converter.

Kafka does not define a record format. This guide uses a LoRaWAN-style record from a network server: the device EUI, the frame port, frequency, RSSI, and four sensor readings packed as hex bytes in data:

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

1. Go to Integrations ⇾ Data converters and click + Add data converter ⇾ Create new converter.

2. Converter type — keep Uplink.

3. Integration type — select Kafka.

4. Name — enter a name, for example Kafka Uplink Converter.

5. Main decoding configuration — the editor opens with the generic default decoder, which expects deviceName in the payload. Replace it with the decoder below. It uses EUI as the device name, reads one byte of data per reading, and copies the radio fields to telemetry. The editor opens in TBEL; use the TBEL / JS toggle to switch to JavaScript.

   • TBEL (Recommended)
   • JavaScript

   /** 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,
   

   ▼Expand

   To adapt:

   • Other byte layout — change the substring positions. For a multi-byte value, read four or more hex characters, for example hexToInt(payloadJson.data.substring(0, 4)) for a big-endian 16-bit value.
   • Scaling — divide or multiply a reading, for example hexToInt(payloadJson.data.substring(0, 2)) / 10.
   • Device properties — move fields such as serial from telemetry to attributes.
   • Server time — remove ts to store records with the time ThingsBoard receives them.
   • Other payload — for a flat JSON record with deviceName and deviceType, keep the generic default decoder.

6. Click Test decoder function and check the result (see Test the Decoder).

7. Click Add.

Decoder Inputs

The decoder signature is function decoder(payload, metadata): object | object[].

The integration does not pass the record key, headers, topic, partition, or offset to the decoder. To tell sources apart, put the information into the record value, or create one integration per topic with its own Metadata.

The integration reads the record value as UTF-8 text. JSON and plain text arrive intact; binary formats such as Avro or Protobuf do not. Encode binary data as hex or Base64 inside a JSON record, as data does in this guide.

Test the Decoder

Click Test decoder function below the editor. Set Payload content type to JSON, replace the payload with the sample record, and click Test. The Output panel shows:

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

Check that deviceName equals EUI and that the four readings match the table above.

1. Go to Integrations ⇾ Integrations and click + Add integration.
2. Basic settings:
   • Integration type — select Kafka.
   • Name — enter a name, for example Kafka integration.
   • Enable integration and Allow create devices or assets — keep both on, so the device is created on its first record.
   • Click Next.
3. Uplink data converter:
   • Click Select existing and choose Kafka Uplink Converter.
   • Alternatively, click Create new to write a decoder here.
   • Click Next.
4. Connection:
   • Group ID and Client ID — enter thingsboard-kafka-group and thingsboard-kafka-client-01.
   • Bootstrap servers — replace YOUR_KAFKA_DOMAIN:9092 with localhost:9092.
   • Topics — replace my-topic-output with my-topic.
   • For a secured broker, expand Advanced settings and add the security properties under Other properties.
   • Leave the other parameters at their defaults. See Connection Settings for details.
5. Click Add.

Advanced settings

Publish a record to my-topic, then confirm that ThingsBoard decoded it and created the device.

Create the integration first. A new consumer group starts from the end of the topic, so it skips records published before its first connection. To read them, add auto.offset.reset = earliest to Other properties and use a new Group ID.

Publish the sample record with the console producer:

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

To check that the record reached the topic, read it back:

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

Publish the test record with the console producer The console consumer prints the record from my-topic

• Integration events — go to Integrations ⇾ Integrations, open the integration, and click the Events tab. The uplink appears with status OK. Click ⋅⋅⋅ in the Message column to see the record.
• Converter events — open Kafka Uplink Converter in Integrations ⇾ Data converters and click the Events tab. In shows the record, Out shows the decoded result, and Metadata shows the integration name.
• Device — go to Devices & assets ⇾ Devices. The integration has created 43T1YH-REE with type Monitoring sensor. Open it and check the eight keys on the Latest telemetry tab: Temperature, Humidity, Fan, Pressure, Port, Freq, rssi, and serial.

Kafka Integration has no downlink. To publish data from ThingsBoard to Kafka, add a kafka rule node to a rule chain. The node publishes the message data as the record value. On success, it adds offset, partition, and topic to the message metadata. For all node settings, see the kafka node reference.

This section publishes shared attribute updates of the device to a separate topic, my-topic-downlink.

1. Create the topic:

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

2. In another terminal, start a consumer and leave it running:

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

Create the my-topic-downlink topic Leave the consumer running to watch my-topic-downlink

1. Go to Data processing ⇾ Rule chains and open the Root Rule Chain.
2. From the External section, drag a kafka node onto the canvas.
3. Set:
   • Name — Publish to Kafka.
   • Topic pattern — my-topic-downlink. The pattern also accepts ${key} placeholders, replaced with values from the message metadata.
   • Bootstrap servers — localhost:9092.
   • Keep the other fields at their defaults. For a secured broker, add the same security properties to Other properties.
4. Click Add.
5. Connect the message type switch node to the kafka node with the Attributes Updated relation.
6. Click Apply changes.

1. Go to Devices & assets ⇾ Devices, select 43T1YH-REE, and open the Attributes tab.
2. Select Shared attributes and click +.
3. Set Key to temperatureUploadFrequency, Type to Integer, Value to 60, and click Add.

• Kafka — the consumer on my-topic-downlink prints the record:

  {"temperatureUploadFrequency":60}

• Rule node events — turn on Debug mode for the kafka node before the test. Its Events tab then shows an IN event with the Attributes Updated relation and an OUT event with the Success relation. The OUT metadata holds offset, partition, and topic.

To read debug events, open the integration or converter, click Events, and select an event: In is the received data, Out is the converter result, and Error holds the error text.

• Integrations overview — how integrations, converters, uplinks, and downlinks fit together.
• Uplink data converter — decoder inputs, output format, and scripting patterns.
• kafka node — publish Rule Engine messages to Kafka.
• Remote Integration — run the integration outside the ThingsBoard cluster.
• TBEL — built-in functions and syntax for converter scripts.
• Kafka consumer configuration — properties you can add to Other properties.