Mqtt publisher plugin

Note: This plugin requires InfluxDB 3 Core.8.2 or later and paho-mqtt>=2.1,<3.

The MQTT Publisher Plugin sends rows out of InfluxDB 3 Core to an MQTT broker. It publishes what is already in the database — it computes nothing — which makes it the outbound half of the pair whose inbound half is mqtt_subscriber.

• Three ways to be driven: a write trigger publishes rows as they land, a scheduled trigger publishes what a table has gained since the previous tick, and an HTTP trigger replays a time range on demand
• Payload shapes consumers expect: JSON, InfluxDB line protocol, or a template you write; one message per row, one per field with the bare value in the body, or one carrying a whole batch
• Topics built from the data: plant/{machine}/{_table} resolves per row, and a substituted value can never add a topic level
• Survives a broker outage: nothing is handed to a disconnected client; a write trigger buffers under a byte budget and a drop policy, while a scheduled trigger stops reading and leaves the rows in the database
• Runs no threads: the connection lives in the trigger’s cache and is serviced by the invocation itself

Contents

• The three triggers — what they share, how they differ
• Common behaviour — the connection and the life of a message
• Write trigger · Scheduled trigger · HTTP trigger — each with its own behaviour, full parameter list and examples
• Topics · Payloads · Row selection
• Where settings come from — layers and formats
• Installation · Observability · Troubleshooting

The three triggers

One plugin file serves all three. They share the connection, the topic and payload machinery, the filters and the buffer; they differ only in where the rows come from and therefore in what can go wrong.

Which one to use

The three can coexist on one database — a write trigger for the live stream and an HTTP trigger for backfills is the common pairing. Give them different client_id values, or none at all, because a broker allows one session per identifier and evicts the previous holder.

Common behaviour

Everything in this section is the same whichever trigger you use.

The connection

A connection is built once and kept in influxdb3_local.cache, so an invocation that already has one spends well under a millisecond publishing. There is no background network loop: the invocation pumps the socket itself, which is what collects acknowledgements and notices a connection that has gone.

An invocation does this, in order:

1. Pump briefly. A pending FIN from the broker is seen before any payload is handed over.
2. Decide whether the client can be trusted. It is replaced — never repaired — if the connection dropped, if a connection setting changed, or if it has been silent longer than max_silence.
3. Drain the buffer, handing over as many messages as the in-flight window and the back-pressure threshold allow.
4. Pump again, collecting whatever acknowledgements arrive within the budget.

Three consequences are worth knowing before configuring anything:

Why replace rather than reconnect. reconnect() leaves paho holding the messages that were in flight and resending them itself, so the plugin could never retire one the broker keeps refusing: a single oversized message would close the connection on every retry and nothing behind it would ever move.

Why no keepalive ping. Proving a quiet link needs a PINGREQ, which paho does not expose. A trigger quiet for longer than max_silence is close to being dropped by the broker anyway, so the plugin pre-emptively builds a new client instead — and on a busy trigger max_silence is never reached.

QoS 0 and silence. With qos = 0 there is no acknowledgement to prove the link is alive, and paho’s PINGRESP is not visible to the plugin, so a trigger invoked less often than max_silence rebuilds its client on each invocation. That is harmless — a fresh client still delivers — only slightly wasteful; raise max_silence or use qos = 1 if the reconnects are unwanted.

The life of one message

row ──filters──▶ topic + payload ──▶ buffer ──▶ publish() ──▶ broker
       │               │               │            │
    skipped        too large      drop_policy    unconfirmed ──ack──▶ confirmed
                   → dropped      max_message_age               │
                                                         reason ≥ 0x80
                                                         → dropped

• Filters are applied first, so nothing is encoded for a row that will not be published.
• Encoding produces the topic and the payload. A message larger than max_payload_bytes is dropped and logged; a batch is split instead.
• The buffer holds what has not been handed over, bounded by max_queue_bytes — counted as payload + topic + 320 bytes of Python overhead, so the limit tracks real memory rather than payload bytes.
• Handing over happens only while the client is connected and the window has room. At QoS ≥ 1 the message waits in unconfirmed until acknowledged.
• A reason code of 0x80 or above (MQTT 5 only) names the message it refers to, so there is nothing to retry: it is dropped at once, counted and logged, without spending max_delivery_attempts. The exception is 0x97 quota exceeded, which is about the rate rather than the message.

There is no delivery guarantee in either direction. A message can be lost: the buffer lives in memory and a restart discards it, QoS 0 packets go with a dropped connection, and drop_policy and max_message_age discard by design. A message can equally be duplicated: an invocation that fails after handing rows over re-reads them, and QoS 1 is at-least-once by definition.

Write trigger

Publishes rows as they are written. The trigger specification may be table:<name> or all_tables; rows of a table not named in source_measurements are ignored either way, so all_tables is safe and is what you want when the buffer should drain on any write.

How it behaves

• The batch is all there is. These rows exist nowhere the plugin can read them back from, so the in-memory buffer is the only thing between a broker outage and lost data. That is the opposite of the scheduled trigger, and it is why max_queue_bytes, drop_policy and max_message_age matter here.
• Ingest is never stalled. An invocation hands over what it can within pump_budget and returns; whatever is left waits for the next write.
• No schema is needed up front. A trigger created before the first write to its table starts with a warning and checks its placeholders and filter types once the table appears.
• A column added later is picked up. On a write trigger the row that brings a new column triggers one schema re-read, so a new tag is emitted as a tag immediately. A scheduled or backfill trigger selects only the columns it already knows, so it picks up a new column within an hour — the schema cache’s TTL — or at once if the trigger is recreated.

What an outage looks like, with max_queue_bytes = 64 MiB, drop_policy = drop_oldest, max_message_age = 5min:

t+0s     broker killed
t+1s     the pump notices; nothing more is handed over at any QoS
t+0..∞   every invocation appends to the buffer and returns as fast as ever;
         once the budget is reached, drop_policy decides what goes
t+7min   broker returns; the next invocation reconnects, drains the buffer
         oldest-first discarding anything past max_message_age, then publishes
         the new rows

Setup

influxdb3 create trigger \
  --database mydb \
  --plugin-filename gh:influxdata/mqtt_publisher/mqtt_publisher.py \
  --trigger-spec "table:press_line" \
  --trigger-arguments 'broker_url=mqtt://broker:1883,source_measurements=press_line,topic=plant/{machine}/{_table}' \
  mqtt_publish_press
influxdb3 enable trigger --database mydb mqtt_publish_press

Examples

Source row used throughout:

press_line,machine=press7,line=l1 temp=21.5,pressure=4.2,state="run" 1791019630123456789

Mirror a table to one topic

--trigger-arguments "broker_url=mqtt://broker:1883,source_measurements=press_line,topic=test4"

test4   {"time":"2026-10-03T09:27:10.123456789Z","machine":"press7","line":"l1",
         "temp":21.5,"pressure":4.2,"state":"run"}

A topic per machine, line protocol — the format to use when the far side is another InfluxDB, because mqtt_subscriber parses it:

--trigger-arguments "broker_url=mqtt://broker:1883,source_measurements=press_line,topic=plant/{machine}/{_table},payload_format=lineprotocol"

plant/press7/press_line   press_line,machine=press7,line=l1 temp=21.5,pressure=4.2,state="run" 1791019630123456789

One topic per field, bare values — the Unified Namespace shape

--trigger-arguments "broker_url=mqtt://broker:1883,source_measurements=press_line,topic=plant/{machine}/{_field},payload_fields=temp pressure,layout=field,retain=true"

plant/press7/temp       21.5
plant/press7/pressure   4.2

retain = true means a dashboard subscribing an hour later sees the current value at once instead of waiting for the next change.

Publish only the rows above a threshold. The template contains commas, which the trigger CLI would split, so this one goes in a TOML file:

source_measurements = ["press_line"]
filters = [["temp", ">=", 185]]
topic = "plant/{machine}/cmd/setpoint"
payload_format = "template"
payload_template = '{"setpoint": 180, "reason": "temp {temp}"}'
qos = 1

plant/press7/cmd/setpoint   {"setpoint": 180, "reason": "temp 187.2"}

Remember that filters selects rows, not transitions: while the temperature stays above 185, every row that arrives is published.

Publishing a value another plugin derives. The trigger is onwrite, not every:…, and that is not a matter of taste: a rollup row carries the timestamp of the start of the bin it summarises and is written once that bin has closed, so a scheduled trigger would find every such row below its watermark.

influxdb3 create trigger --database plant \
  --plugin-filename gh:influxdata/mqtt_publisher/mqtt_publisher.py \
  --trigger-spec "table:press_line_1m" \
  --trigger-arguments "broker_url=mqtt://broker:1883,source_measurements=press_line_1m,topic=plant/{machine}/rollup,payload_fields=temp_avg temp_max" \
  publish_rollup

Announcing availability

--trigger-arguments "broker_url=mqtt://broker:1883,source_measurements=press_line,topic=plant/{machine}/{_table},status_topic=plant/publisher/status"

plant/publisher/status   online     (retained, on connect)
plant/publisher/status   offline    (retained, the will, when the connection dies)

Parameters

Connection

Availability

Source

Row selection

Message

Limits and buffering

Common

Scheduled trigger

Publishes what each source table has gained since the previous tick. Create it without --run-asynchronous: a watermark describes one reader, and two ticks overlapping on a cold cache publish the same first page twice.

How it behaves

There is no lookback setting, because a fixed window is not idempotent: five minutes of lookback on a one-minute schedule publishes every row five times, and duplicates that would collapse inside a table are five real MQTT messages. A watermark per source table defines the range instead:

• bounded below by time > watermark and above by the invocation’s start time less settle_delay;
• a whole page is read into the buffer, and the watermark follows it only once that page has drained — no further page is read while anything is pending;
• on a cold cache the watermark starts at now − settle_delay, so the first tick publishes nothing and the second publishes what appeared since.

Why a page at a time. Handing over is capped by the in-flight window, so a tick can hand over fewer rows than it read. If the watermark followed the last row handed over and those rows fell inside a group sharing one timestamp — 300 machines on aligned seconds produce exactly that — the watermark could not enter the group without abandoning the rest, nor stay below it without re-reading the same rows next tick. The group would republish for ever.

Why a page always ends on a timestamp boundary. The cursor is a timestamp, so a page stopping inside a group would advance past rows that were never published. A page that is entirely one timestamp is re-read whole, up to max_group_rows.

settle_delay exists because the watermark moves along the data’s timestamps, not along the order rows were written. Two lags push a row’s visibility past its timestamp:

• The server’s own write-to-visibility lag. A freshly written row is not queryable until the next WAL flush (up to --wal-flush-interval, default 1s). A tick whose window reaches now reads nothing yet, advances the watermark past that time, and the row is lost once it flushes. This hits every writer, which is why settle_delay defaults to 2s — above the default flush interval. Keep it at or above the server’s --wal-flush-interval.
• A source that batches. It writes a row late relative to the time it carries, and once the watermark has passed that time the row is never published:

12:00:05  tick publishes rows with time ≤ 12:00:05, watermark = 12:00:05
12:00:08  Telegraf flushes a batch timestamped 12:00:00 … 12:00:07
          everything in it at or below 12:00:05 is already behind — lost

Telegraf’s default flush_interval is 10 s. A table another plugin derives is late by that plugin’s whole interval: a minute rollup carries the timestamp of the start of the bin and can only be written once the bin has closed, so no few-second settle_delay ever covers it. Use a write trigger on that table, which has no watermark at all.

The database is the buffer. While the client cannot take anything, the tick reads nothing and leaves the watermark where it is, so an outage costs memory nothing and loses nothing:

t+0s     broker killed
t+1min   the tick connects, fails, logs it and reads nothing; the watermark
         stays where it was
t+1..7m  the same each tick; the buffer stays empty and memory is flat
t+7min   broker returns; each tick publishes up to max_messages per page,
         oldest first, until it has caught up

A tick works through as many pages as tick_budget allows, not one. One page per tick is a ceiling, not a pace: a source writing 18 000 rows a minute against a max_messages of 10 000 never catches up on an every:1min schedule. A tick that stops with a full page still waiting logs a warning.

Filters are compiled into the query, so max_messages counts rows that will actually be published rather than rows scanned.

Setup

influxdb3 create trigger \
  --database mydb \
  --plugin-filename gh:influxdata/mqtt_publisher/mqtt_publisher.py \
  --trigger-spec "every:1min" \
  --trigger-arguments 'broker_url=mqtt://broker:1883,source_measurements=press_line,topic=plant/{machine}/sampled,settle_delay=15s' \
  mqtt_publish_sampled
influxdb3 enable trigger --database mydb mqtt_publish_sampled

Example

A minute’s worth of readings, coalesced into one message per machine, with fifteen seconds of tolerance for a batching source:

source_measurements = ["press_line"]
topic = "plant/{machine}/sampled"
settle_delay = "15s"
layout = "batch"
payload_fields = ["temp", "pressure"]
max_message_age = "10min"
qos = 1

plant/press7/sampled   [{"time":"…:10.123456789Z","temp":21.5,"pressure":4.2},
                        {"time":"…:11.004000000Z","temp":21.6,"pressure":4.2}]

max_message_age must exceed settle_delay, and should exceed settle_delay plus the schedule interval — otherwise each tick skips the part of its window the next tick’s clamp has already passed. The plugin refuses the first; the second it can only log, because the trigger specification never reaches it.

Parameters

Connection

Availability

Source

Row selection

Message

Limits and buffering

Common

HTTP trigger

Replays a time range on demand. One call handles one table and never touches the scheduled watermark, so a backfill and a live scheduled trigger can run against the same table without interfering.

How it behaves

The range is [start, end) — backfill_start inclusive, backfill_end exclusive — so adjacent calls chain without republishing the row on the seam. Both may be omitted, and then the whole table is published, up to max_backfill_messages.

The call works through the range in pages, waiting between them for the unconfirmed total to fall below max_unconfirmed_bytes, so a large range cannot outrun the broker and overflow the buffer.

The answer carries two timestamps, and they are not the same thing:

• reached — the last fully confirmed timestamp group. A group of which some messages are confirmed and some are not cannot be resumed from, so it does not count;
• resume_from — what to pass next time, already adjusted for the direction and the inclusivity of the bound it goes into, so nobody has to do nanosecond arithmetic on an RFC 3339 string.

max_backfill_messages caps the whole call, bounding each page by the budget left so it holds per call, not only between pages, at timestamp-group granularity. Reaching it with rows still to publish is an ordinary stop, not an error: the answer says complete: false and carries a resume_from. Reaching it with the range already exhausted answers complete: true, so an exact hit on the last page costs no spurious resume. That cap is what makes a request with no range safe to issue by accident.

A long backfill outlasts its HTTP response. The plugin finishes regardless and logs progress — there is no way to stop it short of removing the trigger, which is why the cap’s default is low enough to notice.

This trigger ignores status_topic. Its client exists for one request, and its coming and going says nothing about availability.

Setup

influxdb3 create trigger \
  --database mydb \
  --plugin-filename gh:influxdata/mqtt_publisher/mqtt_publisher.py \
  --trigger-spec "request:mqtt_backfill" \
  --trigger-arguments 'broker_url=mqtt://broker:1883,source_measurements=press_line,topic=replay/{machine}/{_table}' \
  mqtt_backfill
influxdb3 enable trigger --database mydb mqtt_backfill

Making a request

curl -X POST "http://localhost:8181/api/v3/engine/mqtt_backfill" \
  -H "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"backfill_start": "2026-10-05T00:00:00Z", "backfill_end": "2026-10-05T06:00:00Z"}'

{
  "status": "ok",
  "table": "press_line",
  "published": 184320,
  "confirmed": 184320,
  "dropped": 0,
  "reached": "2026-10-05T05:59:59.000000000Z",
  "resume_from": "2026-10-05T05:59:59.000000001Z",
  "complete": true
}

A run that stops with rows still to publish — interrupted, or capped with the range not yet exhausted — answers "status": "aborted" with "complete": false; pass its resume_from as backfill_start to continue from exactly where it stopped:

curl -X POST "http://localhost:8181/api/v3/engine/mqtt_backfill" \
  -H "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" -H 'Content-Type: application/json' \
  -d '{"backfill_start": "2026-10-05T05:59:59.000000001Z"}'

Replaying into a different shape. A request may change which rows are taken, where they go and what the message says — which is what a replay for a new consumer needs:

curl -X POST "http://localhost:8181/api/v3/engine/mqtt_backfill?backfill_order=newest_first" \
  -H "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" -H 'Content-Type: application/json' \
  -H 'X-Influxdb3-Mqtt-Publisher-Topic: replay/{machine}/{_table}' \
  -d '{
    "source_measurements": "weld_line",
    "backfill_start": "2026-10-05T00:00:00Z",
    "payload_fields": ["temp", "pressure"],
    "time_format": "ms",
    "filters": [["temp", ">=", 185]]
  }'

What a request may set

Only these, in the body, a header or a query parameter:

backfill_start, backfill_end, backfill_order, source_measurements, filters, allow_values, deny_values, topic, topic_missing_value, payload_format, payload_template, payload_fields, layout, time_format.

Which rows to take, where they go and what the message says stay open because that is what a backfill varies. The broker, the credentials, the QoS, the TLS material and every limit belong to the trigger, and a request naming one is refused — the connection is the trigger’s identity, and a caller who could supply broker_url would be using the plugin to reach a broker of their own.

Parameters

Connection

Source

Row selection

Message

Limits and buffering

Backfill range

Common

Topics

A plain string is valid and is the common case:

topic = test4

A topic is not something that exists

Nothing has to be created in advance, and the plugin creates nothing. In MQTT a topic is just a string in the PUBLISH packet: the broker matches it against whatever subscriptions exist at that moment and delivers it to those. If nobody is subscribed, the message is discarded and that is not an error — under MQTT 5 the broker says so with reason code 0x10 no matching subscribers, which this plugin counts as delivered, because it is.

Three things do constrain which topics you may use:

• The broker’s ACLs. Publishing to a topic the client is not allowed is refused, which is authorisation rather than existence. Under MQTT 5 it comes back as 0x87 and the message is dropped and logged; under 3.1.1 Mosquitto acknowledges it and discards it silently, and nothing here can tell.
• retain = true makes a topic persist. The broker stores the last retained message per topic and hands it to every later subscriber. That is the point of the UNS pattern and of status_topic — and it is also the reason a topic template must produce a bounded set of topics. One topic per machine is a few hundred; one per timestamp would be one per row, stored for ever. That is why {time} is refused in a topic and offered only in the payload.
• Broker-specific rules. Managed brokers cap topic length and depth — AWS IoT Core allows seven forward slashes and 256 bytes — and reserve the $ prefix for their own topics, which this plugin refuses outright.

A placeholder is {, a name matching [A-Za-z_][A-Za-z0-9_]*, and }. The template is scanned for exactly that; every other character, braces included, is literal, so a JSON-shaped template needs no escaping. {{ and }} produce literal braces.

topic = plant/{machine}/line/{line}/{_table}/{_field}

Rules that apply only when a placeholder is used:

• Every placeholder must name a column of at least one source table. The check happens when the schema is first read; requiring it in every table would make a trigger over several unlike tables impossible. A table that does not exist yet cannot be checked at all — the normal order for a write trigger — so that is one warning, and the check waits for the table.
• An empty value is a runtime condition, not an error. In InfluxDB a tag may legitimately be absent from part of a table. Such a row uses topic_missing_value; unset, the row is skipped and counted.
• A substituted value containing /, +, # or a null character is refused. / is refused because a value is data, and data must not be able to add a topic level: a tag value of ../cmd/setpoint would otherwise steer the message into somebody else’s command topic.
• A topic that can never be published is refused with the settings. Empty, starting with $ (the broker’s own namespace) or carrying an empty level — when no placeholder can change that, it is a configuration error rather than a line in the log for every row. A value that renders into one of those shapes is still refused and counted per row.

A column whose name falls outside the grammar — InfluxDB allows spaces and dots in column names — cannot be interpolated, and there is no escape syntax. payload_fields still reaches it; a topic that needs it needs it renamed.

Payloads

A payload carries the timestamp first and the remaining columns in name order, in JSON and in line protocol alike. A write trigger receives a row in the engine’s order and a query returns it in the statement’s, so without one order the same row would encode two ways depending on which trigger published it — a difference that matters to anyone comparing or hashing payloads.

The two interact in two places that would otherwise be guesswork:

• layout = field ignores payload_format. The payload is the value and nothing else, so there is no object to shape and no template to fill. Setting one is not an error, but it has no effect.
• template under batch renders once per row and joins with a newline, the same shape lineprotocol takes in a batch. A JSON-shaped template then produces NDJSON, which a consumer can actually read.

A column empty on a row does not appear in that row’s payload, and a row left with no selected values is skipped. One rule settles two things: a write row simply has no such key while a query returns every column and fills the gaps with None, so without it the same logical row would be published as {"temp":21.5} by one trigger and {"temp":21.5,"pressure":null} by another — and a line protocol payload must carry at least one field.

lineprotocol carries types, taken from the catalog: i for int64, u for uint64, t/f for boolean, quotes for a string. A bare number in line protocol is a float, so an integer published without its suffix would be re-ingested as a float and collide with the column it came from. A Python value cannot tell int64 from uint64 — both are int — which is why the schema is read rather than guessed. Escaping follows the line protocol rules exactly.

payload_fields narrows the payload only. A topic placeholder still resolves against the whole row, so topic=plant/{machine} works with payload_fields=temp.

Row selection

Two mechanisms, deliberately non-overlapping:

• filters — scalar comparisons on column values. It selects rows, not transitions: every row satisfying the condition is published, for as long as it keeps satisfying it. Publishing on the crossing instead is what state_change and threshold_deadman_checks are for — they write an event row, and this plugin publishes that table.
• allow_values / deny_values — membership tests, meant for string tags. The list values are matched as strings, so on a numeric or boolean column a write trigger (str(185.0) == "185.0") and a scheduled or backfill trigger (SQL IN ('185'), coerced by the engine) can disagree. For a numeric or boolean column use filters (temp:'>=185'), which is typed and matches the same rows on every trigger type.

A row that does not carry the column is treated differently by each, and the difference is the one people get wrong:

The two that publish are the ones SQL gets wrong on its own: NULL != 'x' and NULL NOT IN (…) are both NULL rather than TRUE, so a bare != or NOT IN drops exactly those rows. Each negative test therefore compiles with its own IS NULL arm, and both paths — compiled SQL and per-row Python — select the same rows. Without it, adding one deny_values would silently stop publishing every row lacking that tag, which on a table written by more than one source can be most of it.

Filter columns and literals are checked against the schema when it is first read: a column no source table has, an ordering comparison on a text column, or a number compared against a text column are all refused with both names in the message. Values are always bound as query parameters, never written into the statement as text.

Where settings come from

The write and scheduled triggers read three layers; the HTTP trigger reads six:

query parameters > headers > request body > TOML file > trigger arguments > environment

An unrecognised name is refused with the list of accepted ones rather than ignored, in every layer the operator controls — a trigger argument, a key in the TOML file, a body field or a query parameter. A misspelled setting that silently did nothing would be found much later.

config_file_path names a layer rather than setting a value, so it is accepted only from the environment and the trigger arguments: neither the file nor a request can point the plugin at another file.

Two ways to write a setting

A trigger argument can only be a string, so every setting has a compact string form. A TOML file and a JSON request body carry real data structures, and four settings accept those too — which reads better and keeps the types you wrote.

filters also accepts a list of objects — [{"column": "temp", "operator": ">=", "value": 185}] — for a caller that generates the request. In the structured form the value keeps the type you wrote, so 185 is a number; in the compact form, and for a quoted "185", it is read as one.

The compact form’s grammar: pairs are separated by whitespace outside quotes, and quotes say which spaces belong to a value. Either quote character may be the outer pair, so a value carrying one kind is written with the other — state:"=o'brien". There is no escape for a quote inside the same kind; a value needing both belongs in the structured form. One separator then serves both levels — the pairs and the lists inside them — and a decimal literal stays intact, which a dot separator could not: temp:'>=1.5' state:'!=in progress'. An unbalanced quote is a configuration error naming the setting.

Installation

• InfluxDB 3 Core: 3.8.2 or later, with the Processing Engine enabled
• Python packages: influxdata-plugin-utils>=0.5.0, paho-mqtt>=2.1,<3

One virtual environment serves every plugin on a node, so a requirement pinned here can still be overridden by one pinned elsewhere. This plugin needs paho 2.x: it registers CallbackAPIVersion.VERSION2 and reads reason codes from on_publish, neither of which exists in 1.x.

1. Start InfluxDB 3 Core with the Processing Engine enabled:

   influxdb3 serve \
     --node-id node0 \
     --object-store file \
     --data-dir ~/.influxdb3 \
     --plugin-dir ~/.plugins

2. Install the packages:

   influxdb3 install package influxdata-plugin-utils
   influxdb3 install package paho-mqtt

Observability

Counters reach the log as a summary line and the mqtt_publisher_stats table in the trigger’s own database, at most once per stats_interval.

Rows are tagged with broker (host:port) and instance_name.

“Published” is not a delivery count. What it proves depends on the QoS: at QoS 0 nothing about the broker having received it, at QoS 1 that the broker accepted it at least once, at QoS 2 that the four-way handshake completed. PUBCOMP is not a promise of exactly-once here — that needs session state surviving a disconnection, and the plugin connects with a clean session by design. QoS 2 costs two extra round trips and adds nothing to QoS 1.

Code overview

Troubleshooting

Nothing arrives, and the log says nothing is wrong

On a scheduled trigger over a table another plugin writes, check settle_delay. A derived row carries the timestamp of the period it describes and lands a whole interval later, so under the default settle_delay (2s) it is below the watermark by the time it appears and is never published. Use a write trigger on that table, or set settle_delay above the producing plugin’s interval plus its offset and run time.

An empty tick is also indistinguishable from a quiet table — the plugin does not warn about it, because telling the two apart would cost an extra query per tick and the right configuration is known in advance from how the source writes.

On InfluxDB 3 Core.11, a scheduled trigger or a backfill publishes only part of a table

If the table has no tag columns at all, this is a known server defect rather than the plugin: on the upgraded storage engine — the default for new clusters since 3.11.0 — a query carrying a time predicate against a tagless table returns only the rows still in the write buffer, and the ones already flushed to disk are missing with no error. Both query-driven triggers read exactly that way; a write trigger is unaffected because it runs no query.

Fixed in 3.12.0, where a time-filtered query returns every row of a tagless table. On 3.11.x, add any tag column to the table so its rows stay visible to the query.

max_message_age and settle_delay are refused together

max_message_age must exceed settle_delay, or the clamp puts the lower bound of every window above its upper bound and nothing is ever published; with the default settle_delay of 2s, a max_message_age of 2s or less is refused. It should also exceed settle_delay plus the schedule interval, or each tick skips the part of its window the next tick’s clamp has already passed. The plugin cannot check the second, because the trigger specification never reaches it, but the clamp logs every range it skips.

The scheduled trigger falls further and further behind

The source writes more rows per interval than one tick publishes. Raise tick_budget, raise max_messages, or shorten the schedule. A tick that spends its budget with a full page still waiting says so in the log.

The retained status says offline but the publisher is healthy

The trigger is invoked less often than 1.5 × keep_alive, so the broker closes the connection between calls and publishes the will. Either leave status_topic unset, or arrange for the trigger to fire often enough — for a write trigger that means all_tables rather than one sparse table.

The retained status says online after the trigger was deleted

Deleting a trigger does not close its client’s socket; the broker notices when its keepalive expires, 1.5 × keep_alive after the last packet, and publishes the will then. At the default that is seven and a half minutes. Deleting and recreating releases it in about three seconds instead.

A connection attempt blocks the trigger for minutes

connect_timeout bounds the TCP connect and the CONNACK wait, but a TLS handshake and a WebSocket upgrade are bounded by keep_alive — paho’s doing, not the plugin’s — and name resolution by the system resolver alone. Against a TLS endpoint that accepts the connection and then goes silent, lower keep_alive.

At QoS 1 a large burst drains slowly

One invocation confirms only about max_inflight messages per round trip within pump_budget, then returns; the rest stay in the in-memory buffer and drain on the next invocation. At the defaults (max_inflight = 20, pump_budget = 50ms) a single 50 000-row write at qos = 1 confirmed ~50 on the first invocation and climbed ~50 per later write — nothing is lost, but a one-shot burst on a table written rarely delivers slowly. Raising the two together helps dramatically: the same 50 000 rows at pump_budget = 2s, max_inflight = 200 confirmed ~12 000 in one invocation. So for high QoS 1 volume, raise max_inflight and pump_budget (and max_unconfirmed_bytes with them), watch all_tables so any write drains the buffer, or publish at qos = 0, where a whole batch goes out in one invocation. Remember the buffer is in memory: an undrained backlog is lost if the server restarts.

The broker throttles and the plugin does not say why

Under mqtt_version = 5 a quota refusal arrives as 0x97 and is logged, and the invocation stops handing messages over without dead-lettering any of them. MQTT 3.1.1 has no reason codes, so the broker simply closes the connection and it is indistinguishable from any other disconnection — a reason to prefer version 5 against a broker with a published rate limit, AWS IoT Core’s hundred publishes a second per connection being the common one.

A message the broker’s ACL forbids is counted as delivered

Under MQTT 3.1.1 PUBACK carries no reason code, and Mosquitto answers a forbidden publish with an ordinary one rather than closing, so the broker discards the message and the plugin cannot know. Under mqtt_version = 5 the same publish returns 0x87 and is dropped, counted and logged.

Memory grows while the broker is down

That is a write trigger holding what it cannot publish, bounded by max_queue_bytes. Lower it, or set max_message_age so stale readings are discarded rather than queued. A scheduled trigger does not do this — it stops reading instead.

Two triggers keep disconnecting each other

They share an explicit client_id. The broker allows one session per identifier and must evict the previous holder, so give each trigger its own in its own trigger arguments, or point them at separate files with config_file_path — a shared TOML file ranks above the trigger arguments and cannot be overridden from them.

On a cluster every message arrives several times

--node-spec defaults to all, so every process node owns the trigger and runs it: a schedule fires once per owning node on every tick, a write trigger once per owning node on every WAL flush. Pin it with --node-spec nodes:<node-id>. Each node holds its own cache, so its own watermark, buffer and client, and nothing reconciles them.

A trigger created before its table exists logs a warning

That is expected and is the normal order for a write trigger: placeholders and filter types cannot be checked against a schema that does not exist yet. They are checked once the table appears.

Report an issue

For plugin issues, see the Plugins repository issues page.

Logging

Logs are stored in the _internal database (or the database where the trigger is created) in the system.processing_engine_logs table. To view logs:

influxdb3 query --database _internal "SELECT * FROM system.processing_engine_logs WHERE trigger_name = 'your_trigger_name'"

Log columns:

• event_time: Timestamp of the log event
• trigger_name: Name of the trigger that generated the log
• log_level: Severity level (INFO, WARN, ERROR)
• log_text: Message describing the action or error

Report an issue

For plugin issues, see the Plugins repository issues page.

Find support for InfluxDB 3 Core

The InfluxDB Discord server is the best place to find support for InfluxDB 3 Core and InfluxDB 3 Enterprise. For other InfluxDB versions, see the Support and feedback options.