---
pdfexport: true
alias: tutorials-connectiot-kafka
timetoread: true
tutorial: full
module: Connect IoT (Equipment Integration)
description: "This tutorial integrates a plant-wide Apache Kafka event backbone with MES using Connect IoT: consuming production and state events to drive MES, publishing MES confirmations back to Kafka, and handling reliable processing with manual commit and a dead-letter queue"
---

# Kafka Integration Tutorial

This tutorial integrates a plant-wide **Apache Kafka** event backbone with MES using the Connect IoT **Kafka driver**. Unlike an equipment protocol such as SECS/GEM, Kafka does not connect to a single machine: it connects to a shared **event streaming platform** onto which edge gateways, PLC bridges, line controllers, and enterprise systems publish and from which they consume. This makes Kafka the integration layer of choice when many producers and consumers must be decoupled across a site or across the enterprise.

By the end you will have a Connect IoT controller that:

1. **Consumes** production and equipment-state events from Kafka topics and turns them into MES actions (production tracking and resource state for OEE).
2. **Publishes** an MES confirmation event back onto Kafka for downstream consumers (data lake, analytics, other lines).
3. **Processes reliably** using manual offset commit and a **dead-letter queue (DLQ)** for messages MES rejects.
4. **Pauses and resumes** consumption during maintenance windows.

!!! note
    During this tutorial, the Automation Manager will run in console mode in order to highlight the most important events as they take place.

## Apache Kafka Overview

### Overview

Apache Kafka is a distributed event streaming platform used to build real-time data pipelines. Producers write **records** (messages) to named **topics**; consumers read them. Kafka retains records for a configured period, so consumers can read at their own pace and replay history — a good fit for buffering shop-floor events between bursty equipment and slower business logic.

The Connect IoT Kafka driver uses the [@confluentinc/kafka-javascript](https://www.npmjs.com/package/@confluentinc/kafka-javascript) library backed by librdkafka for high-performance native connectivity.

### Key concepts

| Concept | Meaning in a manufacturing integration |
| --- | --- |
| **Topic** | A named event stream, for example `factory.line1.units.completed`. You typically model one topic per event kind per line/area. |
| **Partition** | A topic is split into partitions for parallelism and ordering. Records with the same **key** always land on the same partition, preserving per-key order (for example, all events for one resource). |
| **Offset** | The position of a record within a partition. Committing an offset records "I have processed up to here". |
| **Consumer group** | A set of consumers that share the partitions of the subscribed topics. Connect IoT consumes under a configured group id. |
| **Key / Headers / Value** | A record carries an optional **key** (used for partitioning, for example the resource name), optional **headers** (metadata), and a **value** (the payload, typically JSON). |

Table: Kafka key concepts

### Why Kafka on the shop floor

- **Decoupling** — edge gateways and PLC bridges publish events without knowing MES exists; MES consumes without knowing the producers. Either side can be changed or restarted independently.
- **Buffering** — if MES is briefly unavailable, records accumulate in the topic and are processed on recovery instead of being lost.
- **Fan-out** — one production event can feed MES, a data lake, and an analytics pipeline at once, each with its own consumer group.

## Scenario

Line **LINE1** in a packaging area has an edge gateway that bridges its PLCs to Kafka. The gateway publishes to three topics:

| Topic | Published when | Example value (JSON) |
| --- | --- | --- |
| `factory.line1.units.completed` | A unit finishes at a station | `{"resource":"PACK-01","material":"LOT-4471","quantity":1,"ts":"2026-07-01T10:15:02Z"}` |
| `factory.line1.equipment.state` | A station changes state | `{"resource":"PACK-01","state":"Executing"}` |
| `factory.line1.alarms` | An alarm is raised/cleared | `{"resource":"PACK-01","code":"E-102","text":"Label jam","active":true}` |

Table: Topics published by the LINE1 edge gateway

The record **key** is always the resource name (for example `PACK-01`), so all events for one resource stay ordered within a partition.

The integration goals are:

- **Flow A (consume → MES):** on each `units.completed` record, register production in MES; on each `equipment.state` record, update the resource state so OEE is calculated; on each alarm, log/raise it.
- **Flow B (publish ← MES):** after MES registers production, publish a confirmation to `factory.line1.mes.confirmed` for downstream consumers.
- **Reliability:** if MES rejects a `units.completed` record (for example an unknown resource), route the raw record to `factory.line1.units.completed.dlq` and commit past it so the line is not blocked.

## Prerequisites

- A running Automation Manager and access to the GUI. If you are new to controllers and protocols, complete the [Connect IoT - Basic Configuration Tutorial](scenarios/connectiot_configuration_basic.md) first.
- Reachable Kafka broker(s) and their `hostname:port` addresses (for a local trial, a single-broker cluster on `localhost:9092` is enough).
- The topics above existing on the broker. Create them beforehand, or set `autoCreateTopics` to `true` on the protocol for a trial. Pre-creating topics is recommended for regex subscriptions if you need them discovered immediately, rather than on the next metadata refresh (see the note in *Flow A*).
- A way to produce test records — for example the `kafka-console-producer` CLI shipped with Kafka, the edge gateway itself, or the web UI from the demo environment below.

### Demo environment (containers)

For a self-contained trial you do not need a full cluster. The following `docker-compose.yml` starts a single-node **Apache Kafka** broker (KRaft mode, no ZooKeeper) plus **kafbat/kafka-ui**, a web console for creating topics and inspecting records. The Automation Manager connects to the broker on `localhost:9095`; the UI is served at `http://localhost:8080`. Host port `9095` is used instead of the default `9092` so the demo broker does not clash with an existing broker (for example a MES development `kafka` container) already using `9092`.

```yaml
services:
  kafka-line1:
    image: apache/kafka:3.8.0
    container_name: kafka-line1
    ports:
      - "9095:9092"                      # host 9095 -> container 9092; Automation Manager connects to localhost:9095
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,INTERNAL://:19092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9095,INTERNAL://kafka-line1:19092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,INTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-line1:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"   # matches autoCreateTopics for a trial

  kafka-ui:
    image: kafbat/kafka-ui:latest
    container_name: kafka-ui-line1
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: line1
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka-line1:19092
    depends_on:
      - kafka-line1
```

Start it with `docker compose up -d`, then open `http://localhost:8080` and create the tutorial topics (`factory.line1.units.completed`, `factory.line1.equipment.state`, `factory.line1.alarms`, `factory.line1.mes.confirmed`, and `factory.line1.units.completed.dlq`) under the `line1` cluster.

!!! note
    The service is named `kafka-line1` to avoid clashing with an existing container named `kafka` (for example a MES broker) on the same host or network. If you rename it, update `KAFKA_ADVERTISED_LISTENERS`, `KAFKA_CONTROLLER_QUORUM_VOTERS`, and the UI `BOOTSTRAPSERVERS` to match — they all resolve the broker by its Docker service name.

!!! note
    The broker advertises two listeners: `localhost:9095` for host applications such as the Automation Manager, and `kafka-line1:19092` for the UI on the internal Docker network. Both are needed — a client on the wrong listener cannot connect. On a secured cluster you would swap the `PLAINTEXT` listener for `SASL_SSL`/`mTLS` and set the matching `authenticationMethod` on the protocol.

## Automation Protocol

Create an **Automation Protocol** and select the package `@criticalmanufacturing/connect-iot-driver-kafka`. Configure the connection to your cluster:

| Parameter | Value in this tutorial | Notes |
| --- | --- | --- |
| `bootstrapServers` | `localhost:9095` | Comma-separated broker addresses. **Required.** Use `9095` if you started the demo containers; use `9092` (or your cluster's address) otherwise. |
| `clientId` | `connect-iot-line1` | Identifies this client on the broker; useful when reading broker-side logs. |
| `consumerGroupId` | `mes-line1` | **Required because we consume events.** Keep it stable — Kafka tracks committed offsets per group, so reusing the same id resumes where you left off. |
| `authenticationMethod` | `None` for a local trial | For a secured cluster choose `SASL_Plain`, `SASL_SSL Plain`, `SASL_SSL SCRAM512`, or `mTLS` and fill in the credentials/certificates. See the [Kafka driver reference](../../../userguide/automation/definition/automation-protocol/automation-protocol-protocols/driver_kafka.md#protocol-parameters). |

Table: Protocol connection parameters

For this scenario also review:

- `producerAcks` = `All` (default) — Flow B confirmations are durable (wait for all in-sync replicas). Lower to `Leader` or `None` only for high-volume, loss-tolerant telemetry.
- `fromBeginning` = `false` (default) — this scenario only wants to react to new records going forward, not replay history. This applies consumer-wide, to every event configured below.
- `maxPollIntervalMs` — increase from the default `300000` (5 min) if a single message can trigger long-running MES work, so the broker does not evict the consumer mid-processing.
- `topicMetadataRefreshIntervalMs` — the default (`300000`, 5 min) is fine here since all topics are literal and pre-created; lower it if you switch to the regex event below and want new topics discovered sooner.
- `checkTopicsOnConnect` = `true` — validate the literal topics against the broker during setup and fail fast if the gateway has not created them yet.

!!! note
    Note that to do this in a real-life context, it is important to have general knowledge about the cluster (broker addresses, security, retention, and how topics are named and partitioned) and to agree the topic and payload contract with whoever owns the producers.

## Automation Driver Definition

Create an **Automation Driver Definition** that depends on the Kafka Protocol created above, and select the **Kafka Driver Tasks**. The Kafka driver models the integration through **events** (consume) and **commands** (publish/commit/pause), configured here.

### Configure the consume events

Add one **Event** per topic you consume. The **Event Device Id** is the topic name; it is mandatory and unique.

| Event name | Event Device Id (topic) | Extended data |
| --- | --- | --- |
| Unit Completed | `factory.line1.units.completed` | `topicType: Literal`, `autoCommit: false` (manual — see *Reliability*) |
| Equipment State | `factory.line1.equipment.state` | `topicType: Literal`, `autoCommit: true` |
| Line Alarm | `factory.line1.alarms` | `topicType: Literal`, `autoCommit: true` |

Table: Consume events

For each event, add the **event properties** you need in the workflow. Each property's `propertyType` selects the Kafka field:

| For the *Unit Completed* event add properties | `propertyType` |
| --- | --- |
| `Value` (the JSON payload) | `Value` |
| `Key` (the resource) | `Key` |
| `Topic` | `Topic` |
| `Partition` and `Offset` (needed for manual commit and DLQ) | `Partition`, `Offset` |

Table: Unit Completed event properties

The *Equipment State* and *Line Alarm* events only need `Value` (and optionally `Key`).

!!! note
    Because the protocol's `fromBeginning` is `false`, only records produced after the consumer first subscribes are read, for all three events. Set it to `true` once if you want to replay every topic from its start when bringing the integration online — this affects all events, since it is a driver-level (not per-event) setting.

!!! important
    If you prefer one event that receives all three topics, set `topicType: Regex` and use a pattern such as `^factory\.line1\..*` as the Event Device Id, then read the real topic from a `Topic` property to branch in the workflow. A topic created after the regex subscription is established is only discovered on the next metadata refresh (`topicMetadataRefreshIntervalMs`, five minutes by default) — pre-create the topics if you need them available immediately. Also avoid overlapping a manual-commit event with a catch-all auto-commit event — a single matching auto-commit event commits the offset for the whole record.

### Configure the publish command

Add a **Command** of type **Publish** for Flow B. Add command parameters (classified by `parameterCategory`):

| Command parameter | `parameterCategory` | Purpose |
| --- | --- | --- |
| `topic` | `Topic` | Destination topic — `factory.line1.mes.confirmed`. Exactly one required. |
| `key` | `Key` | Message key — the resource name, to keep confirmations ordered per resource. |
| `value` | `ValueBase` | The confirmation payload (a JSON string built in the workflow). |

Table: Publish command parameters

Add a second **Command** of type **SendToDLQ** with a `Topic` parameter set to `factory.line1.units.completed.dlq`, and a **Commit** command with `Topic` + `Offset` (`partition`, `offset`) parameters — both used in the *Reliability* scenario.

!!! note
    Note that to do this in a real-life context, general knowledge about the payload contract expected by the downstream consumers is required, so the published `value` matches what they parse.

## Automation Controller

Create an **Automation Controller** of `Workflow Type` **Data Flow**, using the Kafka Driver Definition above, and include the **Kafka Driver Tasks** plus the MES tasks you will use (production registration and resource-state change). The entity type used in this tutorial is a `Resource` representing the line/stations.

The controller will hold one workflow per flow. In the page left panel select `Workflows`, then :material-plus:, and add a **Data Flow** workflow for each scenario below.

<!-- DOC-REVIEW: screenshots pending — no images available in the source captures for the Kafka driver GUI (protocol config, event/command config, and each workflow). Capture and insert them here, mirroring the SECS/GEM and Control Flow tutorials. -->

## Flow A — Consume production events and update MES

### Overview

Flow A turns inbound Kafka records into MES actions: production registration for `units.completed`, a resource-state change for `equipment.state`, and alarm handling for `alarms`. It is the **consume → MES** half of the integration and the entry point for the other flows.

### When to Use

- **Production tracking** — an edge gateway or PLC bridge already publishes unit-completion events and you want MES to register them automatically.
- **OEE and state visibility** — equipment state changes arrive on a topic and MES must reflect Executing/Down so availability and OEE are calculated.
- **Event-driven alarms** — shop-floor alarms are streamed and should be raised, cleared, or logged in MES.

### Build the Workflow

Create a Data Flow named, for example, `On Unit Completed - Register Production`.

1. Drag in an **On Equipment Event** task and select the **Unit Completed** event. Its outputs expose the event properties you configured (`Value`, `Key`, `Partition`, `Offset`).
1. The payload arrives as a JSON string in `Value`. Add a converter (or an object/JSON task) to parse it into fields — `resource`, `material`, `quantity`. In Data Flow you reference upstream data with tokens, for example `{{ $onUnitCompleted.Value }}`.
1. Add the MES task that registers production (for example a data-collection / track-out task appropriate to your model), mapping `resource`, `material`, and `quantity` from the parsed payload to its inputs.
1. Wire the MES task's success output onward to Flow B (below) so a confirmation is published only after MES accepts the record.

Repeat the pattern for the other two topics:

- `On Equipment State` → parse `state` → **Change Resource State** MES task, so MES reflects Executing/Down and OEE is calculated. These events use auto-commit, so no explicit commit is needed.
- `On Line Alarm` → parse the alarm → raise/clear it or log it with a **Log Message** task.

### How It Works at Runtime

When a record lands on a subscribed topic, the driver fires the matching **On Equipment Event** task and the workflow branches by event:

```mermaid
flowchart TD
    A([Record on subscribed topic]) --> B{Which event?}
    B -- units.completed --> C[Parse Value JSON]
    C --> D[Register Production in MES]
    D -->|success| E([Continue to Flow B])
    B -- equipment.state --> F[Parse state]
    F --> G[Change Resource State - OEE updated]
    B -- alarms --> H[Raise / clear / log alarm]
```

!!! note
    Because the record key is the resource name, all events for one resource are delivered in order from the same partition, so state transitions are applied in the sequence the line produced them.

### Troubleshooting

- **The event never fires.** Confirm the **Event Device Id** matches the topic name exactly. With the protocol's `fromBeginning: false` only records produced *after* the subscription are read — produce a fresh record, or set `fromBeginning: true` once to replay. Verify the broker is reachable and the topic exists.
- **The payload cannot be parsed.** `Value` is a JSON string, not an object — add a converter/JSON task before mapping fields. Malformed payloads on *Unit Completed* are handled by the DLQ path in *Reliability*.
- **MES rejects the record** (unknown resource or material). This is expected for bad data — route it through the DLQ as described in *Reliability* so the line is not blocked.

## Flow B — Publish an MES confirmation back to Kafka

### Overview

Continuing from Flow A's success path, Flow B publishes a confirmation event onto `factory.line1.mes.confirmed` so downstream consumers learn that MES accepted a record — without any coupling to MES.

### When to Use

- **Fan-out to downstream systems** — a data lake loader, analytics job, or another line needs the confirmation, each with its own consumer group.
- **Closed-loop acknowledgement** — producers want proof their event was processed by MES.

### Build the Workflow

1. Build the confirmation payload — for example a JSON string `{"resource":"PACK-01","material":"LOT-4471","mesStatus":"Registered","ts":"..."}` — using a converter or object task.
1. Add an **Equipment Command** task targeting the **Kafka** driver and the **Publish** command.
1. Map its inputs:
    - `topic` → `factory.line1.mes.confirmed`
    - `key` → the resource (`{{ ...resource }}`)
    - `value` → the confirmation JSON built above.
1. Optionally override `producerAcks`/`compressionType` on the command; leave them at `UseGlobalSetting` to inherit the protocol defaults.

### How It Works at Runtime

```mermaid
flowchart TD
    A([Flow A: MES registered production]) --> B[Build confirmation JSON]
    B --> C[Equipment Command: Publish]
    C --> D([factory.line1.mes.confirmed])
    D --> E[Downstream consumers: data lake, analytics, other lines]
```

Any Kafka consumer subscribed to `factory.line1.mes.confirmed` now receives the confirmation without any coupling to MES.

### Troubleshooting

- **Nothing is published.** Confirm the **Publish** command is wired to Flow A's MES **success** output, not to the workflow start — a confirmation must only be sent after MES accepts the record.
- **Confirmations arrive out of order downstream.** Set `key` to the resource name so all confirmations for one resource share a partition.
- **A confirmation is lost on broker failover.** Keep `producerAcks` at `All` (the default) so the publish waits for all in-sync replicas.

## Reliability — manual commit and dead-letter queue

### Overview

The *Unit Completed* event was configured with `autoCommit: false` so a record is only committed **after** MES has successfully handled it — giving at-least-once processing instead of losing a record if MES fails mid-processing.

### When to Use

- **No lost production** — the record must survive an MES restart mid-processing and be reprocessed rather than dropped.
- **Poison-message isolation** — a record MES can never accept (unknown resource, malformed payload) must be set aside without blocking the partition behind it.

### Extend the Workflow

Extend the Flow A workflow for *Unit Completed*:

1. **On MES success:** run the **Commit** command with `topic = {{ $onUnitCompleted.Topic }}`, `partition = {{ $onUnitCompleted.Partition }}`, `offset = {{ $onUnitCompleted.Offset }}`. The driver commits `offset + 1`, so the handled record is not re-delivered.
1. **On MES failure** (for example the resource is unknown or the payload is malformed): run the **SendToDLQ** command, passing the original `Value` (and `Key`/headers) so the raw record is preserved on `factory.line1.units.completed.dlq`. Then run the **Commit** command for the same record so the line is not blocked behind a poison message.

An operator or a separate maintenance workflow can later inspect the DLQ topic, fix the root cause, and if appropriate replay the records.

### How It Works at Runtime

```mermaid
flowchart TD
    A([Unit Completed record<br/>autoCommit = false]) --> B[Register Production in MES]
    B -->|success| C[Commit offset+1]
    B -->|failure| D[SendToDLQ raw record]
    D --> E[Commit offset+1]
    C --> F([Record handled, not re-delivered])
    E --> G([Poison message parked, line continues])
```

### Expected Outcomes

| Situation | Commands run | Result |
| --- | --- | --- |
| MES accepts the record | **Commit** | Offset advanced; the record is not re-delivered. |
| MES rejects the record | **SendToDLQ** + **Commit** | Raw record preserved on the DLQ; the line keeps moving. |
| MES fails mid-processing before commit | (none) | Record is re-delivered on restart — at-least-once processing. |

Table: Reliability Outcomes

### Troubleshooting

!!! important
    When multiple events could match the same record, a single matching event with `autoCommit` enabled commits the offset for all of them. Keep the *Unit Completed* topic out of any catch-all/regex auto-commit event, or the manual-commit logic above will be bypassed.

- **Records are reprocessed on every restart.** The **Commit** command is not running (or not on both the success and DLQ paths) — the offset never advances. Confirm `partition` and `offset` are mapped from the same `On Unit Completed` event that is being handled.
- **A bad record blocks the partition.** Ensure the **SendToDLQ** path also runs **Commit**; without it the same poison message is redelivered indefinitely.

## Maintenance — pause and resume consumption

### Overview

During planned downstream maintenance (for example the data lake is offline), you can stop consuming without tearing down the controller.

### When to Use

- **Downstream outage** — a consumer of Flow B confirmations is offline and you want records to accumulate rather than be processed and published into a void.
- **Controlled catch-up** — pause during a maintenance window, then resume and let the buffered records drain.

### Commands

- **Pause one topic:** run a command of type **PauseTopic** with a `topic` parameter, for example `factory.line1.units.completed`.
- **Resume it:** run **ResumeTopic** with the same `topic`.
- **Pause/resume everything:** use **PauseAllTopics** / **ResumeAllTopics**.

Records keep accumulating in the topics while paused and are processed on resume. The pause state is preserved across reconnects, so a broker blip will not silently resume a topic you paused.

## Execution

Run the Automation Manager in console mode and start the controller. Then produce a test record.

Using the **web UI** ([http://localhost:8080](http://localhost:8080) from the demo environment): open the `factory.line1.units.completed` topic, select **Produce Message**, set **Key** to `PACK-01`, and paste the payload below as the **Value**:

```json
{"resource":"PACK-01","material":"LOT-4471","quantity":1,"ts":"2026-07-01T10:15:02Z"}
```

The same record can be produced from the CLI instead:

```bash
kafka-console-producer --bootstrap-server localhost:9095 \
  --topic factory.line1.units.completed \
  --property parse.key=true --property key.separator=:
# then type:  PACK-01:{"resource":"PACK-01","material":"LOT-4471","quantity":1,"ts":"2026-07-01T10:15:02Z"}
```

Observe that:

1. The **On Equipment Event** task fires for *Unit Completed* in the Automation Manager console.
1. The MES production-registration task runs.
1. Flow B publishes to `factory.line1.mes.confirmed` — open that topic in the UI (**Messages** tab) and confirm the confirmation record is there.
1. The **Commit** command commits the offset — in the UI, the `mes-line1` **Consumer group** shows lag return to `0`, and re-running the controller does not reprocess the record.

To exercise the DLQ, produce a record with an unknown resource (for example `"resource":"NOPE"`) and confirm it appears on `factory.line1.units.completed.dlq` in the UI while the `units.completed` consumer group commits past it.

!!! note
    Kafka payloads and low-level diagnostics can be large. If console/file logs grow too much, set `logging[].options.maxLength` for the controller/driver file transports in the Connect IoT `config.json`, and use the `logLowLevelMessages` protocol parameter (low-level client logs are prefixed with `##`) to control verbosity. See the [Kafka driver reference](../../../userguide/automation/definition/automation-protocol/automation-protocol-protocols/driver_kafka.md#logging).

## Summary

In this tutorial you connected Connect IoT to a Kafka event backbone and built a realistic MES integration: consuming production, state, and alarm events to drive MES and OEE (Flow A), publishing MES confirmations back to Kafka for downstream consumers (Flow B), and making processing reliable with manual offset commit and a dead-letter queue. You also saw how to pause and resume consumption for maintenance. Because producers and consumers are decoupled through topics, each side — edge gateways, MES, analytics — can evolve independently, which is what makes Kafka a strong backbone for plant-wide and enterprise manufacturing integration.

For the complete list of parameters, commands, event properties, and runtime behavior, see the [[user-guide-driver-kafka|Kafka driver reference]].