Building Event-Driven Manufacturing Microservices with Redpanda Connect

Modern manufacturing environments generate an immense stream of events from programmable logic controllers (PLCs), supervisory control and data acquisition (SCADA) systems, edge sensors, and quality inspection devices. On the shop floor, Message Queuing Telemetry Transport (MQTT) has become the dominant protocol for moving this telemetry between devices and systems. Its publish-subscribe model lets sensors push data to central brokers without needing to know who consumes it, and its lightweight design keeps memory and CPU usage low enough to run on the most constrained embedded hardware.

Collecting events, however, only solves part of the problem. The data still needs to be interpreted as it arrives. I wanted to see if you could catch a quality breach or a bearing about to fail the moment the reading came in, using nothing but off-the-shelf MQTT and Redpanda Connect and no custom middleware. Batch-oriented systems and periodic ERP polling are fine for a lot of manufacturing use cases, but not for that.

In this tutorial, I'll show you how to build a complete, runnable pipeline that:

1. Simulates sensor telemetry and publishes it via MQTT.
2. Ingests those events into Redpanda through Redpanda Connect, using its built-in MQTT input.
3. Routes enriched events to domain-specific Redpanda topics.
4. Powers three Python microservices (quality SPC, predictive maintenance, and a live dashboard) that consume events and produce alerts.

Prerequisites

Before you begin, make sure you have the following installed and running:

Tool	Details	Check Command
Docker Desktop	Includes Docker Engine and Docker Compose. Make sure it is running before you start	docker compose version
Python	3.10 or later	python3 --version
pip	Comes with Python, used to install project dependencies	pip3 --version

macOS note: Port 5000 is often used by AirPlay Receiver. The dashboard service defaults to port 5050 to avoid this conflict. If you want to use a different port, set HTTP_PORT when starting the dashboard.

Understanding the Role of Microservices in Modern Manufacturing

Traditional monolithic manufacturing execution systems (MES) combine quality control, machine management, job scheduling, and reporting into a single application. When high-frequency telemetry must be processed in milliseconds, these monoliths hit their limits. If you integrate a new sensor on a specific line, you have to fully redeploy the entire application. A bug in the new integration can cascade into unrelated functions.

Microservices address this by decoupling the monolith into independent, specialized services:

• A quality service for measurements and SPC tolerances.
• A maintenance service for predictive analytics on vibration and temperature data.
• A dashboard service for real-time KPIs.

Each is developed, deployed, and scaled independently. The quality team can ship a new surface-inspection algorithm while the maintenance team simultaneously improves a bearing-wear model, with no shared deployment windows and no cross-team regression risk.

These advantages hinge on real-time event processing. When a quality measurement falls outside SPC limits, the system must alert operators and potentially halt the line within seconds, before hundreds of defective parts are produced. Delays of even a few minutes, typical of batch processing or periodic database queries, can result in costly scrap and machine damage.

Redpanda enables real-time event processing, designed for high-throughput, low-latency workloads. MQTT data flows through Redpanda Connect into topics where microservices consume it independently, creating a standardized ingestion path that eliminates point-to-point integrations. New services simply subscribe to the relevant topics.

Why MQTT for Shop-Floor Ingestion

Edge devices and PLCs often have limited computing capacity and memory. MQTT requires minimal resources and runs efficiently on even the most constrained hardware. This is important when thousands of sensors transmit data simultaneously across a factory floor.

MQTT's publish/subscribe model fits event-driven architecture naturally. Sensors publish readings to hierarchical topics such as factory/line1/machine-01/temperature or factory/line1/machine-02/vibration without knowing which services consume the data. Adding a new analytics service doesn't require a sensor reconfiguration because the service simply subscribes to the relevant topic tree.

MQTT's quality-of-service levels guarantee delivery even after temporary outages, which is essential in environments where metal structures and electromagnetic interference frequently disrupt connectivity.

Redpanda Connect provides a built-in MQTT input that pulls telemetry directly into Redpanda topics. Sensor data flows into the central event backbone without additional middleware or custom code, and no intermediate databases or REST APIs slow down the data path. Some deployments use MQTT-native brokers built on Redpanda (e.g., Waterstream) to consolidate broker and streaming layers for multi-factory setups.

Event Modeling and Stream Processing Patterns

Before writing any code, it helps to think about how topics, payloads, and delivery guarantees shape the system.

Designing Topics

Topic structure directly influences performance and maintainability. There are two common strategies:

• Physical hierarchy (factory/line1/machine-01/temperature) is used when services focus on specific equipment or locations. A line-specific dashboard subscribes to factory/line1/# to receive only relevant data.
• Functional domains (quality.measurements, maintenance.telemetry) are used when services focus on business functions. A quality service analyzing defects across all lines subscribes to a single topic.

I used physical hierarchy for MQTT ingestion here since it matches how the simulator already publishes data, then let Redpanda Connect's routing logic fan events out into functional Redpanda topics for the services that consume them.

The JSON format is human-readable and easy to debug. For production environments generating millions of events daily, Avro with Redpanda's Schema Registry offers schema evolution and compact serialization. Before a producer can publish data with a changed schema, the registry validates backward compatibility, preventing deployments that break downstream consumers.

Partition keys: Using machine_id as the message key ensures all events from a given machine land in the same partition, preserving ordering for time-dependent analyses like trend calculations and SPC windows.

Stream Processing Patterns

Pattern	Use Case	Example
Windowed aggregation	Continuous KPI calculation	A five-minute rolling window computes machine utilization for OEE dashboards
Complex event processing	Anomaly detection from correlated streams	Temperature rising above baseline while vibration increases triggers a predictive maintenance alert
Enrichment	Adding business context to raw telemetry	A compacted topic containing machine specifications provides tolerance limits without database lookups

Beyond processing patterns, teams must also decide how events are delivered and acknowledged across the system.

Delivery Semantics Trade-offs

• At-least-once: No events are lost; duplicates are possible. Choose this for monitoring and alerting where a duplicate notification is acceptable, but missing a warning is not.
• Exactly-once: Each event is processed exactly once, at the cost of higher latency and resource consumption. Use this for inventory counts, production totals, or financial calculations where duplicates skew results.
• Event replay: Redpanda retains events with configurable retention (commonly 7–30 days). Engineers can reprocess historical events for root-cause analysis of intermittent failures.

Architecture Overview

The end-to-end pipeline looks like this:

Project Structure

To understand the project structure, clone the repository and look at the layout:

bash git clone https://github.com/See4Devs/event-driven-redpanda.git cd event-driven-redpanda

The repository contains three infrastructure configuration files at the root and four service directories, each with its own Python entry point and dependencies:

| event-driven-redpanda/ |-- docker-compose.yml # Orchestrates Redpanda, Mosquitto, Connect, Console |-- mosquitto.conf # Mosquitto broker config (anonymous access for dev) |-- connect.yaml # Redpanda Connect MQTT-to-Redpanda pipeline | |-- sensor_simulator/ | |-- simulator.py # Publishes fake MQTT telemetry for 3 machines | +-- requirements.txt # paho-mqtt | |-- quality_service/ | |-- service.py # SPC rolling-window breach detection -> alerts | +-- requirements.txt # confluent-kafka | |-- maintenance_service/ | |-- service.py # Vibration threshold anomaly detection -> alerts | +-- requirements.txt # confluent-kafka | +-- dashboard_service/ |-- service.py # Last-known-state HTTP API (Flask) +-- requirements.txt # confluent-kafka, flask |
| :---- |

Here is what each component does:

• docker-compose.yml: Defines the infrastructure: a single-node Redpanda broker, the Redpanda Console web UI, an Eclipse Mosquitto MQTT broker, a Redpanda Connect pipeline container, and a one-shot topic-setup container that creates the six Redpanda topics on startup.
• mosquitto.conf: Minimal Mosquitto configuration that listens on port 1883 with anonymous access (suitable for local development).
• connect.yaml: Declarative Redpanda Connect pipeline. Subscribes to MQTT topic factory/line1/#, enriches payloads with ingestion metadata, and routes events to domain-specific Redpanda topics based on sensor_type using Bloblang expressions.
• sensor_simulator/: Python script that publishes fake telemetry for three machines (machine-01, machine-02, machine-03), each with three sensor types (temperature, vibration, pressure), to MQTT topics following the pattern factory/line1/<machine_id>/<sensor_type>. A five percent anomaly injection probability simulates real-world spikes.
• quality_service/: Consumes temperature and pressure events from quality.measurements, maintains a per-machine, per-sensor rolling window (default 30 readings), and publishes alerts to quality.alerts when a reading breaches the mean +/- 3 sigma threshold (simplified Western Electric Rule 1).
• maintenance_service/: Consumes vibration events from maintenance.telemetry and applies threshold-based anomaly detection: WARNING when the rolling average exceeds 3.0 mm/s, CRITICAL when any single reading exceeds 5.0 mm/s. Publishes alerts to maintenance.alerts.
• dashboard_service/: Consumes from all measurement and alert topics to build a last-known-state view of every machine in memory (the "digital twin" pattern), then serves it over a Flask HTTP API on port 5050.

Building the Pipeline: Step by Step

With the project cloned and the file layout in place, you can now walk through each component from infrastructure to application code.

Install Python Dependencies

From the project root, install the dependencies for all four services:

bash pip3 install -r sensor_simulator/requirements.txt \ -r quality_service/requirements.txt \ -r maintenance_service/requirements.txt \ -r dashboard_service/requirements.txt

This installs paho-mqtt, confluent-kafka, and flask.

Start the Infrastructure

Start all containers in the background:

bash docker compose up -d

This brings up five containers:

Container	Purpose	Exposed Port
redpanda	Kafka-compatible streaming broker	19092 (Kafka API), 18082 (HTTP Proxy), 18081 (Schema Registry)
redpanda-console	Web UI for inspecting topics and messages	8080
mosquitto	MQTT broker for sensor ingestion	1883
redpanda-connect	MQTT-to-Redpanda pipeline	none
topic-setup	One-shot container that creates Redpanda topics, then exits	none

Wait a few seconds for the healthchecks to pass, then verify the topics exist:

bash docker exec redpanda rpk topic list

You should see all six topics:

NAME PARTITIONS REPLICAS dashboard.state 3 1 maintenance.alerts 1 1 maintenance.telemetry 3 1 quality.alerts 1 1 quality.measurements 3 1 sensor.raw 3 1

Note: The sensor.raw topic will remain empty during this tutorial. The simulator only produces temperature, vibration, and pressure readings, all of which are explicitly routed to quality.measurements or maintenance.telemetry. The sensor.raw topic acts as a catch-all for any unrecognized sensor types, ensuring unknown events are captured rather than dropped.

You can also open Redpanda Console in your browser to visually inspect topics and messages as they flow through the system. See the Redpanda Console documentation for a full feature overview.

Understand the Redpanda Connect Pipeline

Before starting the simulator, take a moment to review the connect.yaml file that is already running inside the redpanda-connect container. Here it is in full:

| yaml input: mqtt: urls: - tcp://mosquitto:1883 topics: - "factory/line1/#" client_id: "redpanda-connect-ingest" qos: 1 pipeline: processors: # Step 1: Parse the raw JSON payload - mapping: | root = this # Step 2: Add ingestion metadata - mapping: | root = this root.ingested_at = now() root.pipeline = "redpanda-connect" # Step 3: Route to topic based on sensor_type - mapping: | let st = this.sensor_type.lowercase() root = this meta target_topic = match $st { "temperature" => "quality.measurements", "pressure" => "quality.measurements", "vibration" => "maintenance.telemetry", _ => "sensor.raw", } output: kafka: addresses: - redpanda:9092 topic: ${! meta("target_topic") } key: ${! json("machine_id") } partitioner: fnv1a_hash max_in_flight: 64 |
| :---- |

Here are the key points:

• input.mqtt.topics uses factory/line1/#. The # is MQTT's multi-level wildcard, subscribing to all sensors on line 1 regardless of machine or sensor type.
• Pipeline.processors uses Bloblang mapping expressions. The match expression routes temperature/pressure events to quality.measurements, vibration events to maintenance.telemetry, and everything else to sensor.raw.
• meta target_topic sets a metadata field that the output.kafka.topic field interpolates with ${! meta("target_topic") }.
• output.kafka points at redpanda:9092 because Redpanda is Kafka API-compatible. Standard Kafka clients and Redpanda Connect's Kafka output work directly against Redpanda without modification.

The following diagram shows how MQTT topics flow through the Redpanda Connect pipeline and get routed to their respective Redpanda topics:

Simulate Sensor Telemetry

The sensor simulator publishes fake readings for three machines, each with three sensor types (temperature, vibration, pressure). Run it from the project root:

bash python3 sensor_simulator/simulator.py

Your output will look like this:

Sensor simulator connected to localhost:1883 Publishing every 2.0s for machines: ['machine-01', 'machine-02', 'machine-03'] -> factory/line1/machine-01/temperature: 72.34 celsius -> factory/line1/machine-01/vibration: 1.45 mm_s -> factory/line1/machine-01/pressure: 5.12 bar -> factory/line1/machine-02/temperature: 71.89 celsius

Each message is a JSON payload with this schema:

json { "event_id": "a1b2c3d4-...", "machine_id": "machine-01", "sensor_type": "temperature", "value": 72.34, "unit": "celsius", "timestamp": "2025-01-15T10:30:00.000Z", "line": "line1", "factory": "factory-north" }

Open Redpanda Console and navigate to the Topics tab. You should see messages flowing into quality.measurements and maintenance.telemetry.

Tip: Leave the simulator running in its own terminal window. Open new terminal windows for the following services.

Run the Quality SPC Service

Open a new terminal. This microservice consumes temperature and pressure events from quality.measurements, maintains a rolling window per machine per sensor, and publishes an alert to quality.alerts when a reading breaches the mean +/- 3 sigma threshold (simplified Western Electric Rule 1).

bash python3 quality_service/service.py

Output:

| Quality SPC service started | broker=localhost:19092 consuming: quality.measurements -> alerting: quality.alerts window=30 sigma=3.0 <- temperature | machine-01 | value=72.34 <- pressure | machine-02 | value=5.12 ALERT machine-03/temperature value=92.15 z=3.41 |
| :---- |

When the simulator injects an anomalous spike (five percent probability), the SPC check fires and produces an alert event:

json { "alert_id": "...", "alert_type": "SPC_BREACH", "machine_id": "machine-03", "sensor_type": "temperature", "value": 92.15, "mean": 72.01, "stdev": 1.48, "z_score": 3.41, "sigma_threshold": 3.0, "window_size": 30, "timestamp": "2025-01-15T10:31:05.000Z" }

Run the Predictive Maintenance Service

Open another terminal. This service consumes vibration events from maintenance.telemetry and applies threshold-based anomaly detection:

• WARNING: rolling average exceeds 3.0 mm/s.
• CRITICAL: any single reading exceeds 5.0 mm/s.

I picked 3.0/5.0 mm/s somewhat arbitrarily for this demo, but in a real deployment you'd tune these against your specific machine baselines. In a production system, this calls an ML inference endpoint. The threshold approach keeps the tutorial self-contained.

bash python3 maintenance_service/service.py

Output:

| Maintenance service started | broker=localhost:19092 consuming: maintenance.telemetry -> alerting: maintenance.alerts warn=3.0 mm/s critical=5.0 mm/s <- vibration | machine-01 | value=1.23 mm/s <- vibration | machine-02 | value=1.10 mm/s CRITICAL machine-03 value=5.24 avg=1.45 |
| :---- |

Run the Dashboard Service

Open one more terminal. The dashboard service materializes the last-known state of every machine by consuming all measurement and alert topics, then exposes an HTTP API:

bash python3 dashboard_service/service.py

Once it's running, query it:

| bash # All machines and their latest sensor readings curl -s http://localhost:5050/machines | python3 -m json.tool # Detail for a specific machine curl -s http://localhost:5050/machines/machine-01 | python3 -m json.tool # Recent alerts across all machines curl -s http://localhost:5050/alerts | python3 -m json.tool |
| :---- |

Here's an example response from /machines:

json { "machine-01": { "temperature": { "value": 72.34, "unit": "celsius", "timestamp": "...", "topic": "quality.measurements" }, "vibration": { "value": 1.23, "unit": "mm_s", "timestamp": "...", "topic": "maintenance.telemetry" }, "pressure": { "value": 5.12, "unit": "bar", "timestamp": "...", "topic": "quality.measurements" } }, "machine-02": { "..." }, "machine-03": { "..." } }

This is the "digital twin" pattern: a continuously updated materialized view in memory, rebuilt entirely from the event stream without relying on an external database.

Where Change Data Capture Still Makes Sense

Event-driven microservices excel at processing real-time MQTT streams, but many factories also run established enterprise systems that were not designed for event-driven architecture. ERP, MES, and warehouse management systems (WMS) store their data in relational databases.

Change data capture (CDC) helps bridge this gap. When a work order is released or inventory is adjusted, these changes occur as database transactions. Redpanda Connect supports CDC connectors for PostgreSQL, MySQL, and MongoDB that capture these deltas and publish them as events automatically.

A minimal CDC configuration for PostgreSQL with Redpanda Connect looks like:

yaml input: postgres_cdc: dsn: "postgres://user:pass@erp-db:5432/manufacturing?sslmode=disable" schema: "public" tables: - work_orders - inventory snapshot: true output: kafka: addresses: - redpanda:9092 topic: 'erp.${! meta("table") }' key: ${! json("id") }

Important: CDC should be used sparingly and exclusively for system-of-record deltas from transactional systems. Sensor streams, machine telemetry, and production events must remain event-native. CDC is not suitable for high-frequency time-series data because it is optimized for relational change capture, not continuous streams.

Cleanup

When you are done experimenting, stop the Python services with Ctrl+C in each terminal, then tear down the Docker infrastructure:

bash docker compose down -v

The -v flag removes the named volume (redpanda-data), freeing disk space. Omit it if you want to preserve topic data between runs.

All the code used in this tutorial is available on GitHub.

Conclusion

In this tutorial, you built a sensor-first, event-driven manufacturing pipeline end to end. MQTT telemetry flows from simulated machines through an Eclipse Mosquitto broker, where Redpanda Connect ingests, normalizes, and routes events into domain-specific Redpanda topics. Three independent Python microservices (quality SPC, predictive maintenance, and a live dashboard) consume these events and produce alerts or materialized views. The entire stack runs locally with Docker Compose.

I liked this setup because it took very little custom code to get there. The routing logic lives in a single Bloblang mapping in connect.yaml, and each microservice only has to worry about its own topic. Where legacy systems like ERP or MES aren't yet event-native, CDC connectors bridge the gap without forcing sensor data through the same relational path.