Apache Pulsar Integration

Apache Pulsar Integration connects ThingsBoard to an Apache Pulsar cluster. ThingsBoard subscribes to the topics you configure and receives messages in batches. An uplink data converter transforms each message into device or asset data, including telemetry and attributes.

Use Apache Pulsar Integration when:

Devices or your applications already publish data to Pulsar topics, self-hosted or on a managed service such as StreamNative Cloud.
You want that data in ThingsBoard dashboards and the Rule Engine without a separate bridge service.

The integration is uplink-only: it does not publish messages to Pulsar.

Architecture

Receive data. A producer publishes a message to a Pulsar topic. The integration consumes the topic through a named subscription and receives messages in batches.
Decode the payload. The uplink converter decodes each message and returns the device name, telemetry, and attributes.
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 message.
Acknowledge the batch. After processing a batch, the integration acknowledges all its messages to Pulsar, including messages the converter failed to decode. Pulsar does not redeliver them.

Prerequisites

Access to ThingsBoard Cloud with integrations enabled for your tenant.
Permission to create integrations and data converters.
A Pulsar broker that ThingsBoard can reach. To start a local test broker, see Set Up Pulsar.
For a secured broker: a JWT token for a role that may consume from the topic.

Set Up Pulsar

Start a standalone broker in Docker. A standalone broker runs all Pulsar components in one container and creates topics on first use. With a managed service, skip this step and take the service URL and token from your provider.

docker run -d --name pulsar -p 6650:6650 -p 8081:8080 apachepulsar/pulsar:latest bin/pulsar standalone --advertised-address localhost

The broker accepts clients on pulsar://localhost:6650. --advertised-address localhost lets clients on the host, such as ThingsBoard, reach the broker. The Pulsar HTTP API is mapped to port 8081, because self-hosted ThingsBoard uses 8080. The short topic name my-topic used in this guide resolves to persistent://public/default/my-topic.

Set Up Apache Pulsar Integration

Create the uplink converter, assign it to a new Apache Pulsar integration, and publish a test message.

Create the Uplink Data Converter

The uplink converter receives each message as a byte array, resolves the target device, and returns its telemetry and attributes. Apache Pulsar Integration uses the Generic converter: the decoder is a TBEL or JavaScript script. For converter types and the full decoder reference, see Uplink data converter.

This guide uses the sample message that the Test decoder function dialog provides by default:

{
    "deviceName": "Thermometer",
    "deviceType": "thermostat",
    "serialNumber": "SN-12345",
    "temperature": 25.5
}

Go to Integrations ⇾ Data converters and click + Add data converter ⇾ Create new converter.
Converter type — keep Uplink.
Integration type — select Apache Pulsar.
Name — enter a name, for example Apache Pulsar Uplink Converter.

Main decoding configuration — the editor opens with the default decoder below. It already handles this message format, so keep it. It takes deviceName and deviceType from the message, the time from ts (milliseconds) or timestamp (ISO 8601), stores the keys in attributeKeyList (here serialNumber) as attributes, and stores all other fields as telemetry. When the message has no time field, it uses the server time. The editor opens in TBEL; use the TBEL / JS toggle to switch to JavaScript.

TBEL (Recommended)
JavaScript

// Decode an uplink message from a buffer
// payload - array of bytes
// metadata - key/value object

/** Decoder **/

// decode payload to JSON
var data = decodeToJson(payload);

// --- Device name and type ---
var deviceName = data.deviceName != null ? data.deviceName : 'Unknown Device';
var deviceType = data.deviceType != null ? data.deviceType : 'default';
// var customerName = 'Customer C';
// var groupName = 'thermostat devices';

// use assetName and assetType instead of deviceName and deviceType
// to automatically create assets instead of devices.
// var assetName = 'Asset A';
// var assetType = 'building';

// --- Timestamp parsing ---
var timestamp = -1;
if (data.ts != null) {
    timestamp = data.ts;
} else if (data.timestamp != null) {
    timestamp = new Date(data.timestamp).getTime();
}
if (timestamp == -1) {
    timestamp = Date.now();
}

// --- Telemetry and attributes ---
var telemetry = {};

// Keys from the payload that should be stored as attributes instead of telemetry
var attributeKeyList = ["serialNumber"];

// Add the payload keys that match the attribute key list to attributes
var attributes = {};
for (var i = 0; i < attributeKeyList.size(); i++) {
    var attributeKey = attributeKeyList[i];
    if (data[attributeKey] != null) {
        attributes[attributeKey] = data[attributeKey];
    }
}

// Keys to exclude from telemetry (device info, attributes and non-telemetry fields)
var excludeFromTelemetryList = ["deviceName", "deviceType", "ts", "timestamp"];
excludeFromTelemetryList.addAll(attributeKeyList);

// Parse all remaining JSON fields as telemetry
telemetry.putAll(toFlatMap(data, excludeFromTelemetryList, true));

// Result object with device attributes/telemetry data
var result = {
   deviceName: deviceName,
   deviceType: deviceType,
// customerName: customerName,
// groupName: groupName,
// assetName: assetName,
// assetType: assetType,
};

// Add attributes to the result only if at least one matching key was found
if (attributes.size() > 0) {
   result.put("attributes", attributes);
}

result.put("telemetry", {
   ts: timestamp,
   values: telemetry
});

/** Helper functions 'decodeToString', 'decodeToJson' and 'toFlatMap' are already built-in **/

return result;

// Decode an uplink message from a buffer
// payload - array of bytes
// metadata - key/value object

/** Decoder **/

// decode payload to JSON
var data = decodeToJson(payload);

// --- Device name and type ---
var deviceName = data.deviceName != null ? data.deviceName : 'Unknown Device';
var deviceType = data.deviceType != null ? data.deviceType : 'default';
// var customerName = 'Customer C';
// var groupName = 'thermostat devices';

// use assetName and assetType instead of deviceName and deviceType
// to automatically create assets instead of devices.
// var assetName = 'Asset A';
// var assetType = 'building';

// --- Timestamp parsing ---
var timestamp = -1;
if (data.ts != null) {
    timestamp = data.ts;
} else if (data.timestamp != null) {
    timestamp = new Date(data.timestamp).getTime();
}
if (timestamp == -1) {
    timestamp = Date.now();
}

// --- Telemetry and attributes ---
var telemetry = {};

// Keys from the payload that should be stored as attributes instead of telemetry
var attributeKeyList = ["serialNumber"];

// Add the payload keys that match the attribute key list to attributes
var attributes = {};
for (var i = 0; i < attributeKeyList.length; i++) {
    var attributeKey = attributeKeyList[i];
    if (data[attributeKey] != null) {
        attributes[attributeKey] = data[attributeKey];
    }
}

// Keys to exclude from telemetry (device info, attributes and non-telemetry fields)
var excludeFromTelemetryList = ["deviceName", "deviceType", "ts", "timestamp"].concat(attributeKeyList);

// Parse all remaining JSON fields as telemetry
toFlatMap(data, telemetry, excludeFromTelemetryList);

// Result object with device attributes/telemetry data
var result = {
   deviceName: deviceName,
   deviceType: deviceType,
// customerName: customerName,
// groupName: groupName,
// assetName: assetName,
// assetType: assetType,
};

// Add attributes to the result only if at least one matching key was found
if (Object.keys(attributes).length > 0) {
   result.attributes = attributes;
}

result.telemetry = {
   ts: timestamp,
   values: telemetry
};

/** Helper functions **/

function decodeToString(payload) {
   return String.fromCharCode.apply(String, payload);
}

function decodeToJson(payload) {
   var str = decodeToString(payload);
   return JSON.parse(str);
}

function toFlatMap(obj, result, excludeList, prefix) {
    prefix = prefix || '';
    for (var key in obj) {
        if (excludeList.indexOf(key) !== -1) continue;
        var value = obj[key];
        var fullKey = prefix ? prefix + '.' + key : key;
        if (typeof value === 'object' && value !== null && !Array.isArray(value)) {
            toFlatMap(value, result, excludeList, fullKey);
        } else {
            result[fullKey] = value;
        }
    }
}

return result;

To adapt:

Device name in another field — replace data.deviceName and data.deviceType, for example with data.deviceId.
Other device properties — add their keys to attributeKeyList, for example firmwareVersion.
Assets instead of devices — uncomment assetName and assetType and use them instead of deviceName and deviceType.
Customer or group — uncomment customerName or groupName to assign the device to a customer or an entity group.
Update-only keys — in Advanced decoding parameters, add keys to the Update only keys list to store them only when their value changes, for example serialNumber.

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

Decoder Inputs

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

Variable	Type	Content
`payload`	byte array	The message body, unchanged. Binary formats arrive intact.
`metadata`	`{[key: string]: string}`	`integrationName` and the key–value pairs from the integration’s Metadata field.

The integration does not pass the topic, message key, or message properties to the decoder. To tell sources apart, put the information into the message, or create one integration per topic with its own Metadata.

Test the Decoder

Click Test decoder function below the editor. The Payload already holds the sample message. Click Test. The Output panel shows:

{
    "deviceName": "Thermometer",
    "deviceType": "thermostat",
    "attributes": {
        "serialNumber": "SN-12345"
    },
    "telemetry": {
        "ts": 1792072084561,
        "values": {
            "temperature": 25.5
        }
    }
}

The message has no time field, so ts is the time of the test.

Create the Integration

Go to Integrations ⇾ Integrations and click + Add integration.
Basic settings:
- Integration type — select Apache Pulsar.
- Name — enter a name, for example Apache Pulsar integration.
- Enable integration and Allow create devices or assets — keep both on, so the device is created on its first message.
- Click Next.
Uplink data converter:
- Click Select existing and choose Apache Pulsar Uplink Converter.
- Alternatively, click Create new to write a decoder here.
- Click Next.
Connection:
- Service URL — replace pulsar://YOUR_PULSAR_DOMAIN:6650 with pulsar://localhost:6650.
- Topics and Subscription name — keep my-topic and my-subscription.
- Credentials type — keep Anonymous for the local broker. For a secured broker, select Security token and paste the token.
- Leave the other parameters at their defaults. See Connection Settings for details.
Click Add.

Connection Settings

Parameter	Description
Service URL	Broker address: `pulsar://host:6650`, or `pulsar+ssl://host:6651` for TLS. With TLS, the integration does not verify the broker certificate. Required.
Topics	Topics to consume, separated by commas without spaces. A short name resolves to `persistent://public/default/<name>`; for another tenant or namespace, enter the full name. Required.
Subscription name	Name of the Pulsar subscription. Pulsar stores the read position per subscription, so the integration resumes from the same position after a restart. Required.
Max number of messages	Maximum number of messages in one batch. Default: `1000`.
Max number of bytes	Maximum size of one batch, in bytes. Default: `10485760` (10 MB).
Timeout in milliseconds	Maximum time to wait for a batch to fill. When it expires, the integration processes the messages received so far. Default: `100`.
Credentials type	Anonymous — no authentication. Security token — JWT authentication with the token in the Token field.
Execute remotely	Runs the integration as a separate process outside the ThingsBoard cluster. ThingsBoard generates an Integration key and Integration secret for it. See Remote Integration.

Advanced settings

Parameter	Description
Description	Free-text description.
Metadata	Key–value pairs added to the `metadata` of every message from this integration.

The subscription is Exclusive: only one consumer may use it. Give each integration that reads the same topic its own Subscription name.

Test Uplink

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

Send a Test Uplink

Create the integration first. A new subscription starts from the latest message, so it skips messages published before it was created. After that, Pulsar keeps unread messages for the subscription while the integration is offline.

Publish the sample message. The pulsar-client commands read it from a file, because the --messages option splits text at commas.

echo '{"deviceName":"Thermometer","deviceType":"thermostat","serialNumber":"SN-12345","temperature":25.5}' | \
  docker exec -i pulsar sh -c 'cat > /tmp/msg.json && bin/pulsar-client produce my-topic -f /tmp/msg.json'

'{"deviceName":"Thermometer","deviceType":"thermostat","serialNumber":"SN-12345","temperature":25.5}' | docker exec -i pulsar sh -c "cat > /tmp/msg.json && bin/pulsar-client produce my-topic -f /tmp/msg.json"

If Pulsar is installed locally, run from the Pulsar directory:

echo '{"deviceName":"Thermometer","deviceType":"thermostat","serialNumber":"SN-12345","temperature":25.5}' > /tmp/msg.json
bin/pulsar-client produce my-topic -f /tmp/msg.json

For a remote broker, add --url pulsar://your-broker-host:6650 before produce.

Install the client with pip install pulsar-client, then run:

import json
import pulsar

client = pulsar.Client("pulsar://localhost:6650")
producer = client.create_producer("my-topic")
producer.send(json.dumps({
    "deviceName": "Thermometer",
    "deviceType": "thermostat",
    "serialNumber": "SN-12345",
    "temperature": 25.5
}).encode("utf-8"))
client.close()

Verify the Result

Integration events — go to Integrations ⇾ Integrations, open the integration, and click the Events tab. The uplink appears with status OK. The integration records the message body as Base64.
Converter events — open Apache Pulsar Uplink Converter in Integrations ⇾ Data converters and click the Events tab. In shows the message, Out shows the decoded result, and Metadata shows the integration name.
Device — go to Devices & assets ⇾ Devices. The integration has created Thermometer with type thermostat. Open it and check temperature on the Latest telemetry tab and serialNumber on the Attributes tab.

Troubleshooting

Symptom	Cause	Fix
“Host must be in the public network.” under Service URL	The server does not allow integrations to connect to private addresses, for example on ThingsBoard Cloud.	Use a public broker address, or a Remote Integration.
Integration does not start	Wrong Service URL, the broker is down, or a firewall blocks the port.	Check the URL scheme and port (`pulsar://…:6650` or `pulsar+ssl://…:6651`) and that the broker accepts connections from the ThingsBoard host.
Same, on a secured broker	The token is missing, expired, or its role may not consume from the topic.	Select Security token and paste a valid token. Check the role’s permissions on the namespace.
Same, with a second integration on the topic	Both use the same Subscription name, and the subscription allows one consumer.	Give each integration its own Subscription name.
No messages received	The producer writes to another topic, or the messages were published before the subscription was created.	Check the full topic name, including tenant and namespace. Publish a new message.
Only the first topic works	Topics contains spaces after the commas.	Separate the topics with commas only.
Device not created	Allow create devices or assets is off, or `deviceName` is empty.	Turn the option on. Check Out in the converter events for `deviceName`.
Decoder error	The message is not JSON, or a field is missing.	Open the converter’s Events tab and read the Error column. The failed message is already acknowledged, so publish a corrected one.

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.