Skip to main content

mqtt_v5

EXPERIMENTAL

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

Pushes messages to an MQTT 5 broker.

# Common config fields, showing default values
output:
  label: ""
  mqtt_v5:
    urls: [] # No default (required)
    client_id: ""
    connect_timeout: 30s
    clean_start: true
    session_expiry_interval: 0s
    topic: "" # No default (required)
    qos: 1
    write_timeout: 3s
    retained: false
    metadata:
      exclude_prefixes: []
    max_in_flight: 64

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

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

Refused publications​

At QoS 1 and above the server answers each publication with a reason code, and a refusal — 0x87 not authorized, 0x97 quota exceeded, 0x90 topic name invalid — is returned as an error carrying that code. There is no setting here for what to do about it: a fallback output puts the error text into a fallback_error metadata field, which a switch can route on for a different destination per reason code.

output:
  fallback:
    - mqtt_v5:
        urls: [ tcp://localhost:1883 ]
        topic: events
    - switch:
        cases:
          - check: 'meta("fallback_error").contains("0x97")'
            output:
              retry:
                output:
                  mqtt_v5:
                    urls: [ tcp://localhost:1883 ]
                    topic: events
          - output:
              file:
                path: ./dead-letter.jsonl

Note that 0x10 "no matching subscribers" is a success, not a failure: it means the message was accepted and nobody was listening.

Metadata as user properties​

Every metadata field on a message is sent as an MQTT 5 user property, except those excluded by the metadata setting. This is the part MQTT 3.1.1 cannot do at all, and it is what carries an identifier or a routing key alongside the payload rather than inside it.

The MQTT 5 properties themselves are separate fields on this output. They arrive from an input as metadata, so unless the field is named here they are not sent as properties at all:

output:
  mqtt_v5:
    urls: [ tcp://destination:1883 ]
    topic: ${! meta("mqtt_topic") }
    content_type: ${! meta("mqtt_content_type") }
    correlation_data: ${! meta("mqtt_correlation_data") }
    # The mqtt_ fields describe the delivery an mqtt_v5 input received
    # rather than the message, so a further hop leaves them behind.
    metadata:
      exclude_prefixes: [ mqtt_ ]

The same applies to response_topic, message_expiry_interval and payload_format_indicator.

Naming one is safe even where a message does not carry it: reading an absent metadata field yields the text null, which leaves the property unset rather than failing the message — so a value that is genuinely those four characters is not sent either. The retained_interpolated field reads an absent value the same way, leaving the configured retained value standing. A value that is present but malformed is still an error, because that is a configuration mistake.

Performance​

This output benefits from sending multiple messages in flight in parallel for improved performance. You can tune the max number of in flight messages (or message batches) with the field max_in_flight.

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"

topic​

The topic to publish messages to. This field supports interpolation functions.

Type: string

qos​

The QoS value to set for each message. Has options 0, 1, 2. At QoS 0 the server sends no acknowledgement at all, so a message it would have refused is indistinguishable from one it accepted.

Type: int
Default: 1

write_timeout​

The maximum amount of time to wait to write data before the attempt is abandoned.

Type: string
Default: "3s"

# Examples

write_timeout: 1s

write_timeout: 500ms

retained​

Set message as retained on the topic.

Type: bool
Default: false

retained_interpolated​

Override the value of retained with an interpolable value, this allows it to be dynamically set based on message contents. The value must resolve to either true or false. This field supports interpolation functions.

Type: string

content_type​

The MIME type describing the payload, sent as the MQTT 5 content type property. This field supports interpolation functions.

Type: string

# Examples

content_type: application/json

content_type: ${! meta("content_type") }

response_topic​

The topic a reply to this message should be published to, sent as the MQTT 5 response topic property. This field supports interpolation functions.

Type: string

correlation_data​

Data a requester uses to match a reply with its request, sent as the MQTT 5 correlation data property. This field supports interpolation functions.

Type: string

message_expiry_interval​

How long, in seconds, the server should keep the message for a subscriber that is not connected. Must resolve to a whole number of seconds. This field supports interpolation functions.

Type: string

# Examples

message_expiry_interval: "300"

message_expiry_interval: ${! meta("expiry_seconds") }

payload_format_indicator​

Whether the payload is UTF-8 text: 1 if it is, 0 if it is unspecified bytes. Must resolve to 0 or 1. This field supports interpolation functions.

Type: string

metadata​

Which metadata values are sent as MQTT 5 user properties. Everything is sent except the prefixes listed here.

When bridging from the mqtt_v5 input, exclude mqtt_: that is the namespace the input writes to describe a message it received — its topic, its QoS, its delivery flags — which describes the hop the message just made rather than the message.

metadata:
  exclude_prefixes: [ mqtt_ ]             # the input's delivery bookkeeping
  exclude_prefixes: [ mqtt_, secret_ ]    # and a prefix of your own

Metadata can also be shaped before it reaches here with a mapping processor, which is the way to rename a field, drop one, or promote one into the payload.

Type: object

metadata.exclude_prefixes​

Metadata keys beginning with any of these are not sent.

Type: array
Default: []

max_in_flight​

The maximum number of messages to have in flight at a given time. Increase this to improve throughput.

Type: int
Default: 64