Apache Pulsar Integration

Apache Pulsar Integration connects ThingsBoard to an Apache Pulsar cluster. ThingsBoard subscribes to a Pulsar topic, consumes incoming messages, and decodes them via an uplink converter into device telemetry and attributes.

Before creating the integration, prepare the following:

• A running Pulsar broker. Use a local Docker installation or a managed service such as StreamNative Cloud.
• A topic to consume from.
• A security token if your broker requires authentication.

The uplink converter decodes incoming Pulsar messages and maps them to the ThingsBoard data model. Apache Pulsar uses a generic uplink converter.

The decoder function receives:

• payload — raw Pulsar message body as a UTF-8 byte array
• metadata — integrationName and any key-value pairs configured in the integration’s Metadata settings

Sample payload:

{
  "deviceName": "Sensor A1",
  "deviceType": "thermostat",
  "temperature": 23.5,
  "humidity": 60
}

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 Apache Pulsar and select Apache Pulsar from the list.
5. Name — enter a converter name, for example Apache Pulsar Uplink Converter.
6. Main decoding configuration — paste the decoder function from the appropriate tab below, then continue with step 7.

The decoder function used in this tutorial:

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

/** Decoder **/

var data = decodeToJson(payload);

var deviceName = data.deviceName != null ? data.deviceName : 'Unknown Device';
var deviceType = data.deviceType != null ? data.deviceType : 'default';

// Timestamp: use ts (Unix ms), timestamp (ISO 8601), or server time
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(); }

var telemetry = {};
var attributes = {
  integrationName: metadata['integrationName']
};

var excludeFromTelemetryList = ["deviceName", "deviceType", "ts", "timestamp"];
telemetry.putAll(toFlatMap(data, excludeFromTelemetryList, true));

var result = {
  deviceName: deviceName,
  deviceType: deviceType,
  attributes: attributes,
  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 **/

var data = decodeToJson(payload);

var deviceName = data.deviceName != null ? data.deviceName : 'Unknown Device';
var deviceType = data.deviceType != null ? data.deviceType : 'default';

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(); }

var telemetry = {};
var attributes = {
  integrationName: metadata['integrationName']
};

var excludeFromTelemetryList = ["deviceName", "deviceType", "ts", "timestamp"];
toFlatMap(data, telemetry, excludeFromTelemetryList);

var result = {
  deviceName: deviceName,
  deviceType: deviceType,
  attributes: attributes,
  telemetry: { ts: timestamp, values: telemetry }
};

/** Helper functions **/

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

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

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;

7. Optionally, click Test payload decoder to validate the script against a sample payload.
8. Click Add.

To adapt this converter to your device:

• Different device name / type fields — replace data.deviceName and data.deviceType with the field names your device sends.
• Timestamp — name the field ts for Unix milliseconds or timestamp for ISO 8601. Without either, the converter falls back to server receive time.
• Telemetry fields — all JSON fields not in excludeFromTelemetryList are mapped to telemetry automatically. Add field names to the list to exclude them.
• Static attributes — add device properties (e.g. firmware version, location) to the attributes object.
• Assets instead of devices — use assetName / assetType instead of deviceName / deviceType.

1. Go to Integrations center ⇾ Integrations and click + Add integration.
2. Basic settings:
   • Set Integration type to Apache Pulsar and enter a name (e.g. Apache Pulsar Integration).
   • Leave Enable integration and Allow create devices or assets on so the device is created automatically on first data.
   • Click Next.
3. Uplink data converter:
   • Click Select existing and choose the Apache Pulsar Uplink Converter created in the previous step, or click Create new to define the decoder inline.
   • Click Next.
4. Connection:
   • Fill in Service URL, Topic, Subscription name, and the remaining parameters.
   • Select Credentials type and provide a token if required.
   • See Connection settings for a full description.
   • Click Add to save the integration.

Basic settings — set Integration type to Apache Pulsar, enter a name, and click Next. Uplink data converter — Create new is selected by default. Paste your decoder function into the editor and click Next. Connection — enter the Service URL, Topic, and Subscription name. Set Credentials type to Anonymous for open brokers or Security token for authenticated clusters.

Service URL

The Pulsar broker endpoint. Use pulsar:// for plain connections (default port 6650) and pulsar+ssl:// for TLS-encrypted connections (default port 6651).

Examples:

• Plain: pulsar://YOUR_PULSAR_DOMAIN:6650
• TLS / managed: pulsar+ssl://broker.example.com:6651

Topic

The Pulsar topic to consume messages from (e.g. my-topic). Pulsar topics follow the format {persistent|non-persistent}://{tenant}/{namespace}/{topic-name}. The broker resolves a short name like my-topic to persistent://public/default/my-topic on a standalone instance.

Subscription name

A string identifier for this consumer subscription (e.g. my-subscription). The subscription tracks the last acknowledged message offset and resumes from the same position after a restart. Use a consistent name across integration restarts.

Max number of messages

Maximum number of messages to fetch in a single poll. Default: 1000.

Max number of bytes

Maximum total payload size to fetch per poll, in bytes. Default: 10485760 (10 MB).

Timeout in milliseconds

Consumer receive timeout — how long the client waits for a message before returning an empty result. Default: 100.

Credentials type

Execute remotely

When enabled, ThingsBoard generates an Integration key and Integration secret that allow the integration to run as a separate process outside the ThingsBoard cluster — useful when the broker is only reachable from a restricted network. See Remote Integration.

After the integration is created, send a test uplink and confirm that ThingsBoard received the message, decoded it correctly, and provisioned the device.

Publish a test message to the Pulsar topic using the pulsar-client CLI tool.

docker exec pulsar bash -c 'printf "%s" "{\"deviceName\":\"Sensor A1\",\"deviceType\":\"thermostat\",\"timestamp\":\"2026-04-09T13:25:00Z\",\"temperature\":23.5,\"humidity\":60}" > /tmp/msg.txt && bin/pulsar-client produce my-topic -f /tmp/msg.txt'

bin/pulsar-client produce my-topic \
  --num-produce 1 \
  --messages '{"deviceName":"Sensor A1","deviceType":"thermostat","temperature":23.5,"humidity":60}'

pip install pulsar-client

import pulsar, json

client = pulsar.Client('pulsar://localhost:6650')
producer = client.create_producer('my-topic')
payload = json.dumps({
    "deviceName": "Sensor A1",
    "deviceType": "thermostat",
    "temperature": 23.5,
    "humidity": 60
})
producer.send(payload.encode('utf-8'))
client.close()
print("Message sent.")

Go to Integrations center ⇾ Integrations, open Apache Pulsar Integration, and check the Events tab. An Uplink event with Status: OK confirms the message was processed. Click … in the Message column to inspect the raw payload.

Go to Integrations center ⇾ Data converters, open the uplink converter, and click the Events tab. Inspect the In (raw payload), Out (decoded result), and Metadata columns to verify the decoder output.

Go to Entities ⇾ Devices. The device Sensor A1 is automatically provisioned on the first message. Open it and check the Latest telemetry tab — temperature and humidity should reflect the published values.

• 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
• Downlink Data Converter — full encoder function reference: input parameters, output fields, and scripting patterns
• Remote Integration — run the integration outside the ThingsBoard server to reach Apache Pulsar from a private network
• TBEL scripting reference — built-in functions and operators for writing converter scripts
• Rule Engine — how the Root Rule Chain routes shared attribute updates to the downlink converter