Batch-Collect MQTT Telemetry into Files on a Schedule

Streaming telemetry does not have to mean streaming infrastructure. This blueprint runs io.kestra.plugin.mqtt.Subscribe on a 15-minute schedule against telemetry/#: each run connects, collects messages until it hits 500 records or 30 seconds of listening, writes the batch to a file in Kestra's internal storage, and posts the collected count to Slack. The result is micro-batching over MQTT with two explicit bounds, so every execution finishes quickly and every batch is a durable, downloadable artifact ready for a downstream load.

How it works

1. collect_telemetry (io.kestra.plugin.mqtt.Subscribe) connects to the broker from the MQTT_SERVER secret and subscribes to the telemetry/# wildcard, matching every device and metric under the telemetry namespace.
2. The task collects until one of two bounds is hit: maxRecords: 500 caps the batch size, and maxDuration: PT30S caps the listening window. Whichever fires first ends the task, so a quiet topic never blocks the execution and a busy one never produces an unbounded batch.
3. Collected messages land as a file in internal storage, exposed as {{ outputs.collect_telemetry.uri }}, with {{ outputs.collect_telemetry.messagesCount }} carrying the batch size. Each stored message includes its topic, payload, QoS, and retain flag.
4. notify posts the count to Slack; the errors block posts a distinct alert when a window fails, so a telemetry gap is a message, not a surprise.
5. A disabled-by-default Schedule trigger runs the collection every 15 minutes.

What you get

• Micro-batching over MQTT with explicit size and time bounds instead of a long-running consumer to babysit.
• Every batch stored as a file output, ready to feed a warehouse load, an object storage upload, or a transformation task.
• A per-window count in Slack, making silent drops in telemetry volume visible as a trend.
• A failure alert that distinguishes a broken broker connection from a merely quiet topic.

Who it's for

• IoT teams landing device telemetry in a warehouse without running a dedicated streaming stack.
• Platform engineers who prefer bounded, replayable batch executions over always-on consumers.
• Anyone who needs MQTT traffic captured as inspectable files before committing to a schema.

Why orchestrate this with Kestra

A bare MQTT consumer script needs a process supervisor, its own batching logic, somewhere durable to write, and monitoring on top. Kestra collapses all of that: the Subscribe task is the batching logic, internal storage is the durable sink, the schedule is the supervisor, and every window is an execution with logs, outputs, and history. Changing the cadence or the bounds is a one-line YAML edit.

Prerequisites

• An MQTT broker reachable from Kestra with traffic on telemetry/#, or adjust the topic filter. Public test brokers accept anonymous connections; for authenticated brokers, add the username and password properties from secrets.
• A Slack incoming webhook for digests and failure alerts.

Secrets

• MQTT_SERVER: broker URI, e.g. tcp://broker.example.com:1883.
• SLACK_WEBHOOK_URL: Slack incoming webhook URL.

Quick start

1. Add the MQTT_SERVER and SLACK_WEBHOOK_URL secrets to your Kestra namespace.
2. Publish a few test messages, e.g. mosquitto_pub -t telemetry/dev-001/temp -m '{"value": 21.4}', then execute the flow within 30 seconds.
3. Check the Slack digest and download the batch file from the execution's Outputs tab.
4. Set disabled: false on the every_15_minutes trigger.

How to extend

• Feed {{ outputs.collect_telemetry.uri }} into a downstream task to load each batch into DuckDB, Postgres, or object storage.
• Tune maxRecords and maxDuration to your traffic: tighter bounds mean fresher data, looser bounds mean fewer executions.
• Narrow the topic filter per flow, e.g. one collector for telemetry/+/temperature and another for telemetry/+/vibration.
• Swap the schedule for the MQTT polling trigger from this plugin when you want collection to start only when messages are flowing.

Links

• MQTT plugin documentation
• MQTT topic wildcards explained