MQTT Source Node

The mqttsource node subscribes to an MQTT broker (MQTT v5) and emits each received message as a record into the pipeline. Incoming messages are buffered in a bounded in-memory queue and drained in batches, so the source streams continuously until the pipeline is terminated.

The source uses the Eclipse Paho MQTT v5 client. It connects to the broker, subscribes to a topic filter, and authenticates with an optional username/password credential. Anonymous brokers are supported by omitting the credential.

Key Features

• MQTT v5 subscription: subscribes to a topic filter supporting + (single level) and # (multi level) wildcards
• Configurable QoS: supports QoS 0 (at most once), 1 (at least once), and 2 (exactly once)
• TLS support: connect over ssl:// or wss:// when useTls is enabled
• Automatic reconnect: reconnects and re-subscribes automatically after a dropped connection
• Bounded receive queue: buffers inbound messages in a bounded queue; drops messages with a warning when the queue is full
• Infinite streaming: never exhausts — runs until the pipeline is terminated

Configuration

Field	Type	Required	Default	Description
brokerUrl	String	Yes	—	MQTT broker URL, e.g. tcp://broker:1883. Must use ssl:// or wss:// when useTls is true
topicFilter	String	Yes	—	Topic filter to subscribe to. Supports + and # wildcards, e.g. sensors/#
clientId	String	Yes	—	MQTT client identifier used for the connection
encodingType	String	Yes	—	Message format. See Encoding Types
credentialId	String	No	—	ID of a username/password credential in jobContext.otherProperties. Omit for anonymous brokers
qos	int	No	1	Quality of Service level. Must be 0, 1, or 2
cleanStart	boolean	No	true	MQTT v5 clean-start flag set on connect
sessionExpiryIntervalSeconds	Long	No	—	MQTT session expiry interval in seconds. Omit for no explicit expiry
useTls	boolean	No	false	Enable TLS. Requires brokerUrl to start with ssl:// or wss://
automaticReconnect	boolean	No	true	Reconnect automatically after a dropped connection
receiveTimeoutMs	Long	No	1000	How long a fetch waits for the first message before returning (milliseconds). Must be positive
maxBatchSize	Integer	No	500	Maximum number of messages returned per fetch. Must be positive
receiveQueueCapacity	Integer	No	10000	Capacity of the bounded inbound queue between the MQTT callback and the framework loop. Must be positive

Authentication

For brokers that require authentication, store a username/password credential in jobContext.otherProperties and reference it by credentialId. The credential's username and password are passed to the MQTT connection. For anonymous brokers, omit credentialId.

DAG Example

jobContext:
  otherProperties:
    mqtt-cred:
      username: fleak
      password: super-secret
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "mqttsource"
    config:
      brokerUrl: "ssl://broker.example.com:8883"
      topicFilter: "sensors/#"
      clientId: "fleak-mqtt-client"
      encodingType: "JSON_OBJECT"
      credentialId: "mqtt-cred"
      qos: 1
      useTls: true
      automaticReconnect: true
      maxBatchSize: 500
      receiveTimeoutMs: 1000
      receiveQueueCapacity: 10000
    outputs:
      - "sink"

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

Related Nodes

• kafkasource: Alternative stream source for Kafka topics
• sqssource: Stream source for Amazon SQS queues