Skip to main content

mqtt_v5

EXPERIMENTAL

This component is experimental and therefore subject to change or removal outside of major version releases.

Subscribe to topics on MQTT 5 brokers.

# Common config fields, showing default values
input:
  label: ""
  mqtt_v5:
    urls: [] # No default (required)
    client_id: ""
    connect_timeout: 30s
    clean_start: true
    session_expiry_interval: 0s
    topics: [] # No default (required)
    auto_replay_nacks: true

This input speaks MQTT 5 only, and will not fall back to 3.1.1. Use the mqtt input for servers that speak the older protocol.

This input uses https://github.com/eclipse/paho.golang, which is itself experimental.

Durable subscriptions​

Receiving messages published while the pipeline was down needs three settings together: clean_start: false, a fixed client_id, and a session_expiry_interval longer than the outage. Any one of them left at its default loses the messages.

Shared subscriptions​

A shared subscription is an ordinary topic filter of the form $share/<group>/<filter>, so it needs no setting of its own. Scaling out means more instances of the same configuration using the same group.

Acknowledgement​

Messages are acknowledged to the server once the pipeline has finished with them, and strictly in the order they arrived, because that is the only order MQTT 5 allows. A message still being worked on therefore holds back the acknowledgement of everything received after it.

A message that never succeeds stops consumption altogether: nothing behind it is acknowledged either, and once receive_maximum messages are outstanding — or the server's own limit, if that field is unset — the server stops sending. Nothing is lost, and the run is redelivered on the next connection, but the pipeline reports itself healthy while consuming nothing. A warning is logged the first time a message is rejected.

So route messages that cannot succeed to a dead-letter destination rather than rejecting them, with a fallback output:

output:
  fallback:
    - kafka_franz:
        seed_brokers: [ localhost:9092 ]
        topic: events
    - file:
        path: ./dead-letter.jsonl

Pipelines that need messages processed in order should also set pipeline.threads: 1.

Metadata​

This input adds the following metadata fields to each message:

- mqtt_topic
- mqtt_qos
- mqtt_retained
- mqtt_duplicate
- mqtt_message_id
- mqtt_content_type (if set)
- mqtt_response_topic (if set)
- mqtt_correlation_data (if set)
- mqtt_message_expiry_interval (if set)
- mqtt_payload_format_indicator (if set)
- mqtt_subscription_identifier (if set)

MQTT 5 user properties are added under their own keys with no prefix, which is the part MQTT 3.1.1 cannot carry. They are written before the fields above, so a publisher cannot overwrite mqtt_topic by sending a user property of that name; where MQTT 5 permits a repeated key, the last occurrence wins.

The mqtt_ fields describe the delivery rather than the message. When publishing on to another MQTT server, exclude that prefix with the mqtt_v5 output's metadata setting; other outputs carry them like any other metadata.

You can access these metadata fields using function interpolation.

Fields​

urls​

A list of URLs to connect to. If an item of the list contains commas it will be expanded into multiple URLs.

Type: array

# Examples

urls:
  - tcp://localhost:1883

client_id​

An identifier for the client connection. Under MQTT 5 this is the key the server stores a session against, so a client that wants to resume a session must present the same identifier every time it connects.

Type: string
Default: ""

dynamic_client_id_suffix​

Append a dynamically generated suffix to the specified client_id on each run of the pipeline. This can be useful when clustering Bento producers. Note that a generated suffix produces a new client identifier on every run, and therefore a new session: combining it with clean_start: false will not resume anything.

Type: string

OptionSummary
nanoidappend a nanoid of length 21 characters

connect_timeout​

The maximum amount of time to wait in order to establish a connection before the attempt is abandoned.

Type: string
Default: "30s"

# Examples

connect_timeout: 1s

connect_timeout: 500ms

keepalive​

Max seconds of inactivity before a keepalive message is sent.

Type: int
Default: 30

user​

A username to connect with.

Type: string
Default: ""

password​

A password to connect with.

Secret

This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info.

Type: string
Default: ""

tls​

Custom TLS settings can be used to override system defaults.

Type: object

tls.enabled​

Whether custom TLS settings are enabled.

Type: bool
Default: false

tls.skip_cert_verify​

Whether to skip server side certificate verification.

Type: bool
Default: false

tls.enable_renegotiation​

Whether to allow the remote server to repeatedly request renegotiation. Enable this option if you're seeing the error message local error: tls: no renegotiation.

Type: bool
Default: false
Requires version 1.0.0 or newer

tls.root_cas​

An optional root certificate authority to use. This is a string, representing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

Secret

This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info.

Type: string
Default: ""

# Examples

root_cas: |-
  -----BEGIN CERTIFICATE-----
  ...
  -----END CERTIFICATE-----

tls.root_cas_file​

An optional path of a root certificate authority file to use. This is a file, often with a .pem extension, containing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

Type: string
Default: ""

# Examples

root_cas_file: ./root_cas.pem

tls.client_certs​

A list of client certificates to use. For each certificate either the fields cert and key, or cert_file and key_file should be specified, but not both.

Type: array
Default: []

# Examples

client_certs:
  - cert: foo
    key: bar

client_certs:
  - cert_file: ./example.pem
    key_file: ./example.key

tls.client_certs[].cert​

A plain text certificate to use.

Type: string
Default: ""

tls.client_certs[].key​

A plain text certificate key to use.

Secret

This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info.

Type: string
Default: ""

tls.client_certs[].cert_file​

The path of a certificate to use.

Type: string
Default: ""

tls.client_certs[].key_file​

The path of a certificate key to use.

Type: string
Default: ""

tls.client_certs[].password​

A plain text password for when the private key is password encrypted in PKCS#1 or PKCS#8 format. The obsolete pbeWithMD5AndDES-CBC algorithm is not supported for the PKCS#8 format. Warning: Since it does not authenticate the ciphertext, it is vulnerable to padding oracle attacks that can let an attacker recover the plaintext.

Secret

This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info.

Type: string
Default: ""

# Examples

password: foo

password: ${KEY_PASSWORD}

clean_start​

Whether to discard any session the server is holding for this client_id when connecting for the first time. Reconnections made by this component are always made with clean start disabled, so that a session established once is resumed rather than thrown away.

A durable subscription is the combination of three settings: clean_start: false, a fixed client_id, and a session_expiry_interval above zero. Any one of them left at its default is enough to lose messages sent while the pipeline was down.

Type: bool
Default: true

session_expiry_interval​

How long the server keeps this client's session after the connection closes. Zero means the session ends the moment the connection does. Rounded down to whole seconds, which is the resolution MQTT 5 carries it at.

Type: string
Default: "0s"

# Examples

session_expiry_interval: 1h

session_expiry_interval: 24h

receive_maximum​

The number of QoS 1 and QoS 2 messages this client is willing to have in flight to it at once. On the input this is how far ahead of the pipeline the server may read; on the output the only inbound traffic it could bound is acknowledgements. Left unset, the server chooses.

Type: int

maximum_packet_size​

The largest packet, in bytes, this client is willing to accept. Left unset, no limit is sent and the server may deliver a packet of any size it supports.

Type: int

will​

Set last will message in case of Bento failure

Type: object

will.enabled​

Whether to enable last will messages.

Type: bool
Default: false

will.qos​

Set QoS for last will message. Valid values are: 0, 1, 2.

Type: int
Default: 0

will.retained​

Set retained for last will message.

Type: bool
Default: false

will.topic​

Set topic for last will message.

Type: string
Default: ""

will.payload​

Set payload for last will message.

Type: string
Default: ""

will.delay_interval​

How long the server waits after the connection drops before publishing the will message. A delay longer than the time it takes the pipeline to reconnect means a brief network outage publishes no will at all. Rounded down to whole seconds.

Type: string
Default: "0s"

reconnect_backoff​

How long to wait between attempts to re-establish a dropped connection. The wait grows from min towards max, and is reset once a connection succeeds.

Type: object

reconnect_backoff.min​

How long to wait before the first reconnection attempt.

Type: string
Default: "1s"

reconnect_backoff.max​

The longest this component will wait between reconnection attempts.

Type: string
Default: "1m"

topics​

A list of topic filters to consume from. A filter of the form $share/<group>/<filter> is a shared subscription, splitting the stream across every instance using the same group.

Type: array

qos​

The level of delivery guarantee to enforce. Has options 0, 1, 2. At QoS 0 the server neither redelivers nor waits for an acknowledgement, so a message lost in transit stays lost.

Type: int
Default: 1

no_local​

Whether to ask the server not to deliver back messages this same client published. Applied to every filter.

Type: bool
Default: false

retain_as_published​

Whether messages forwarded by the server keep the retained flag they were published with. Applied to every filter.

Type: bool
Default: false

retain_handling​

Whether the server should send retained messages when this subscription is made: 0 always, 1 only if the subscription is new, 2 never. Applied to every filter.

Type: int
Default: 0

ack_interval​

How often acknowledgements finished by the pipeline are flushed to the server. A longer interval sends fewer, larger batches; a shorter one returns capacity to the server sooner.

Type: string
Default: "50ms"

on_connect_refused​

What to do when the server refuses the connection outright, which it reports with a reason code such as 0x86 bad user name or password, 0x87 not authorized, or 0x8A banned.

  • retry reconnects for ever, logging the reason code each time. A refusal caused by a missing permission then recovers on its own once the permission is granted, with no restart.
  • fail stops the input instead. Prefer it for batch and one-shot pipelines, where a run that hangs is worse than a run that fails.

The reason code the server sent is logged either way, because the client library's own error text does not carry it.

Type: string
Default: "retry"
Options: retry, fail.

on_subscribe_refused​

What to do when the server refuses one or more of the topic filters, which it reports with a reason code such as 0x87 not authorized or 0x8F topic filter invalid.

A refusal does not close the connection, so without one of these the pipeline would sit connected and healthy while receiving nothing at all.

  • retry subscribes again on a growing delay, for ever, logging each refusal. A filter refused because a permission is missing then starts working on its own once the permission is granted, with no restart.
  • fail stops the input. Prefer it for batch and one-shot pipelines, where a run that hangs is worse than a run that fails.
  • continue carries on with whichever filters were granted. Use it when one filter of several is expendable — but note that the pipeline then runs while knowingly missing that data.

Every refused filter is logged with its own reason code whichever is chosen.

Type: string
Default: "retry"
Options: retry, fail, continue.

auto_replay_nacks​

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.

Type: bool
Default: true