MQTT Trigger

MQTT Trigger
Certified
Poll MQTT topics on a schedule

Polls MQTT topics every interval (60s default) and starts one execution only when at least one message is read. Collects messages up to maxRecords or maxDuration, stores them in internal storage (use {{ trigger.uri }} and {{ trigger.messagesCount }}), and defaults to JSON deserialization with QoS 1; prefer RealtimeTrigger for per-message executions.

yaml
type: io.kestra.plugin.mqtt.Trigger

Examples

yaml
id: mqtt_trigger
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.payload }}"

triggers:
  - id: trigger
    type: io.kestra.plugin.mqtt.Trigger
    server: tcp://localhost:1883
    clientId: kestraProducer
    topic:
      - kestra/sensors/cpu
      - kestra/sensors/mem
    serdeType: JSON
    maxRecords: 10

Properties

topic *object

Topics to consume

String or list of strings; when multiple topics are provided they are all subscribed.

allowConcurrent boolean

Defaultfalse

Specifies whether a trigger is allowed to start a new execution even if a previous run is still in progress.

authMethod string

The Authentication Method

clientId string

A client identifier that is unique on the server being connected to

connectionTimeout string

The connection timeout

crt string

CA certificate for TLS connections

httpsHostnameVerificationEnabled booleanstring

Disable ssl verification

interval string

DefaultPT1M

Formatduration

Interval between polling.

The interval between 2 different polls of schedule, this can avoid to overload the remote system with too many calls. For most of the triggers that depend on external systems, a minimal interval must be at least PT30S. See ISO_8601 Durations for more information of available interval values.

maxDuration string

The max duration waiting for new rows

It's not an hard limit and is evaluated every second

maxRecords integerstring

The max number of rows to fetch before stopping

It's not an hard limit and is evaluated every second

mqttVersion string

DefaultV5

Possible Values

V3V5

The MQTT version to use

password string

The password to use for the connection

qos integerstring

Default1

Sets the quality of service for this message

Quality of Service 0: indicates that a message should be delivered at most once (zero or one times). The message will not be persisted to disk, and will not be acknowledged across the network. This QoS is the fastest, but should only be used for messages which are not valuable - note that if the server cannot process the message (for example, there is an authorization problem). Also known as "fire and forget".
Quality of Service 1: indicates that a message should be delivered at least once (one or more times). The message can only be delivered safely if it can be persisted, so the application must supply a means of persistence using MqttConnectOptions. If a persistence mechanism is not specified, the message will not be delivered in the event of a client failure. The message will be acknowledged across the network.
Quality of Service 2: indicates that a message should be delivered once. The message will be persisted to disk, and will be subject to a two-phase acknowledgement across the network. The message can only be delivered safely if it can be persisted, so the application must supply a means of persistence using MqttConnectOptions. If a persistence mechanism is not specified, the message will not be delivered in the event of a client failure. If persistence is not configured, QoS 1 and 2 messages will still be delivered in the event of a network or server problem as the client will hold state in memory. If the MQTT client is shutdown or fails and persistence is not configured then delivery of QoS 1 and 2 messages can not be maintained as client-side state will be lost.

serdeType string

DefaultJSON

Possible Values

STRINGJSONBYTES

Serializer / Deserializer used for the payload

server string

The address of the server to connect to, specified as a URI

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of execution states after which a trigger should be stopped (a.k.a. disabled).

username string

The user name to use for the connection

when string

Defaulttrue

A condition that determines whether the trigger should run.

A Pebble expression evaluated at trigger time. The trigger fires only when the expression evaluates to a truthy value (true, a non-empty string, a non-zero number). Use this to gate trigger execution on dynamic runtime values such as execution labels, flow variables, or environment conditions.

Outputs

messagesCount integer

Number of messages consumed

Lower than maxRecords when the subscription ended on maxDuration or on a graceful worker shutdown.

uri string

Formaturi

URI of the internal storage file