Sparkplug B Source Node

The sparkplugb_source node subscribes to an MQTT broker and consumes messages that follow the Eclipse Sparkplug B specification used in industrial IoT. It decodes the Google-Protobuf Sparkplug B payloads, resolves each metric to its declared name and type, and emits one flattened record per Sparkplug message into the pipeline.

Sparkplug B is an MQTT topic and payload convention (Eclipse Tahu). Devices publish on topics of the form spBv1.0/{group}/{message_type}/{edge_node}[/{device}]. This source subscribes to a single Sparkplug group (plus the shared STATE topic) and turns the binary payloads into structured records, handling the Sparkplug session lifecycle — births, deaths, sequence numbers, and metric aliases — on your behalf.

Key Features

• MQTT Sparkplug B decode: connects over MQTT 5 (HiveMQ client) and decodes Sparkplug B protobuf payloads for a single Sparkplug group
• Alias and datatype resolution: caches the metric names, aliases, and datatypes declared in NBIRTH/DBIRTH messages, then resolves compact NDATA/DDATA metrics (which may carry only an alias, or have datatypes stripped) back to fully named, typed data points
• Session integrity with automatic rebirth: tracks the per-edge-node Sparkplug sequence number; on a sequence gap or an unresolvable alias it drops the message and publishes a Node Control/Rebirth command (NCMD) to ask the edge node to re-announce its metadata
• Lifecycle awareness: processes NDEATH/DDEATH to mark nodes and devices offline; optionally passes host-application STATE (online/offline) messages downstream
• Backpressure handling: buffers inbound messages in a bounded queue and reports standard source metrics (bytes in, events out, deserialize failures)

How It Decodes Sparkplug B

Sparkplug B organizes traffic by message type on the MQTT topic:

• NBIRTH / DBIRTH (node/device birth): sent when an edge node or device comes online. They carry the full set of metrics with their names, numeric aliases, and datatypes. The source records this metadata and emits the birth as a record.
• NDATA / DDATA (node/device data): ongoing telemetry. To save bandwidth these often reference metrics by alias only and may have datatypes stripped. The source resolves each metric's name and type from the cached birth metadata before emitting it.
• NDEATH / DDEATH (node/device death): the node/device went offline. The source clears the corresponding cached state and does not emit a record.
• STATE (host application status): a retained MQTT message describing whether a Sparkplug host application is online. Emitted as a record when passThroughStateMessages is enabled.

Every payload carries a monotonically increasing sequence number (seq, 0–255). If the source sees a gap — indicating a missed message — or a metric alias it cannot resolve, it drops the affected message and requests a rebirth so the edge node republishes its NBIRTH/DBIRTH. Repeat rebirth requests for the same node are suppressed for rebirthCooldownMs.

Configuration

The full config object for the Sparkplug B source node (SparkplugBSourceDto.Config):

Field	Type	Required	Default	Description
brokerUrl	String	Yes	—	MQTT broker URL, e.g. tcp://host:1883 or ssl://host:8883
groupId	String	Yes	—	The single Sparkplug group ID this instance subscribes to
clientId	String	No	—	MQTT client ID. If blank, a random ID is generated
username	String	No	—	MQTT authentication username
password	String	No	—	MQTT authentication password
tlsEnabled	boolean	No	false	Whether to use TLS (ssl://) for the broker connection
queueCapacity	int	No	10000	Bounded inbound queue size between the MQTT callback and the framework loop. Must be > 0
rebirthCooldownMs	long	No	5000	Per-node cooldown window (ms) suppressing repeat rebirth requests. Must be >= 0
passThroughStateMessages	boolean	No	true	Emit host-application STATE (online/offline) messages as records
cleanStart	boolean	No	true	MQTT clean-start flag on connect
sessionExpirySeconds	long	No	0	MQTT session expiry interval in seconds (0 = expire on disconnect). Must be >= 0
offerTimeoutMs	long	No	1000	Max time (ms) to wait offering to a full inbound queue before dropping a message. Must be >= 0
shutdownGraceMs	long	No	5000	On shutdown, how long (ms) to wait for the queue to drain before forcing exit. Must be >= 0

Output Records

The node emits JSON records with a flattened structure. Telemetry and birth messages (NBIRTH, DBIRTH, NDATA, DDATA) produce:

Field	Type	Description
group_id	String	Sparkplug group ID from the topic
edge_node_id	String	Edge node ID from the topic
device_id	String | null	Device ID for device-level messages; null for node-level
message_type	String	One of NBIRTH, DBIRTH, NDATA, DDATA
seq	Integer | null	Sparkplug sequence number of the payload
bdSeq	Long	Birth/death sequence for the edge node, from the last NBIRTH
timestamp	Long | null	Payload timestamp in epoch milliseconds
received_at_ms	Long	When the source received the MQTT message (epoch ms)
metrics	Array	Resolved metrics (see below)

Each element of metrics is an object:

Field	Type	Description
name	String	Resolved metric name (from alias if the payload only carried an alias)
type	String	Sparkplug datatype, e.g. Int32, Float, Boolean
value	any | null	The metric value (null when the metric is marked null)
timestamp	Long | null	Per-metric timestamp in epoch milliseconds
is_historical	Boolean	Present and true only when the metric is flagged historical
is_null	Boolean	Present and true only when the metric value is null

STATE records have a different shape:

Field	Type	Description
message_type	String	Always STATE
host_id	String	Host application ID from the topic
received_at_ms	Long	When the source received the message (epoch ms)
online	Boolean	Host application online flag, parsed from the STATE payload
timestamp	Long	Host application timestamp from the STATE payload

DAG Example

jobContext:
  otherProperties:
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "sparkplugb_source"
    config:
      brokerUrl: "tcp://localhost:1883"
      groupId: "FactoryA"
      clientId: "zephflow-sparkplugb"
      passThroughStateMessages: true
    outputs:
      - "sink"

  - id: "sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"

The following example authenticates over TLS and only emits telemetry (STATE messages are dropped):

dag:
  - id: "source"
    commandName: "sparkplugb_source"
    config:
      brokerUrl: "ssl://broker.example.com:8883"
      groupId: "plant1"
      username: "mqtt_user"
      password: "mqtt_secret"
      tlsEnabled: true
      passThroughStateMessages: false
      queueCapacity: 20000
    outputs:
      - "sink"

  - id: "sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"

Related Nodes

• kafkasource: Consume streaming data from Kafka topics
• sqssource: Read messages from an Amazon SQS queue