---
alias: user-guide-driver-kafka
tags:
    - connect iot
description: "This documentation details the Kafka Driver for IoT connectivity, outlining data types, protocol parameters, event handling, commands, and runtime behavior"
---

# Connect IoT - Kafka Driver

Apache Kafka is a distributed event streaming platform used for building real-time data pipelines and streaming applications. This driver provides Kafka transport primitives for Connect IoT controllers, enabling message consumption, publishing, offset management, and dead-letter queue handling.

The driver uses the [@confluentinc/kafka-javascript](https://www.npmjs.com/package/@confluentinc/kafka-javascript) library, which provides a KafkaJS-compatible API backed by librdkafka for high-performance native Kafka connectivity.

## Data Types

These are the supported specific data types:

| Name     | Description             |
| -------- | ----------------------- |
| String   | String value            |
| Boolean  | Boolean value           |
| Integer  | Integer value           |
| DateTime | DateTime value          |
| Decimal  | Decimal value           |
| Object   | Structured value (JSON) |

Table: Kafka specific data types

## Protocol Parameters

The protocol supports the following parameters:

| Name                           | Type     | Possible Values                                                   | Default Value            | Description                                                                                                                                                        |
| ------------------------------ | -------- | ----------------------------------------------------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| bootstrapServers               | String   |                                                                   | _(required)_             | Comma-separated Kafka broker addresses (hostname:port) for connecting to the Kafka cluster.                                                                        |
| clientId                       | String   |                                                                   | connect-iot-driver-kafka | Kafka client identifier used when connecting to the broker.                                                                                                        |
| authenticationMethod           | Enum     | None<br>SASL_Plain<br>SASL_SSL Plain<br>SASL_SSL SCRAM512<br>mTLS | None                     | Kafka authentication mode.                                                                                                                                         |
| userName                       | String   |                                                                   |                          | Username for SASL authentication.                                                                                                                                  |
| password                       | Password |                                                                   |                          | Password for SASL authentication.                                                                                                                                  |
| caPem                          | Text     |                                                                   |                          | PEM-encoded CA certificate for SSL/TLS connections. Accepts a file path or the certificate contents.                                                               |
| certificatePem                 | Text     |                                                                   |                          | PEM-encoded client certificate for mTLS authentication. Accepts a file path or the certificate contents.                                                           |
| keyPem                         | Text     |                                                                   |                          | PEM-encoded client private key for mTLS authentication. Accepts a file path or the key contents.                                                                   |
| validateCertificates           | Boolean  |                                                                   | true                     | Validate TLS certificates. Set to false for self-signed certificates.                                                                                              |
| messageMaxBytes                | Integer  |                                                                   | 1048576                  | Maximum Kafka protocol request message size, in bytes. The default is 1MB.                                                                                         |
| autoCreateTopics               | Boolean  |                                                                   | false                    | Allow Kafka auto topic creation on producer and consumer.                                                                                                          |
| logLowLevelMessages            | Enum     | None<br>Info<br>Debug<br>Warning<br>Error                         | Info                     | Minimum low-level Kafka client log level routed through the driver logger. Logs are prefixed with `##`. Use `None` to disable these logs.                          |
| consumerGroupId                | String   |                                                                   | connect-iot-kafka-group  | Kafka consumer group identifier for managing message consumption. Required when events are configured.                                                             |
| fromBeginning                  | Boolean  |                                                                   | false                    | Use the earliest retained offset when a partition has no committed consumer-group offset. Existing committed offsets always take precedence. Applies consumer-wide, to every literal and regex event subscription. |
| maxPollIntervalMs              | Integer  |                                                                   | 300000                   | Maximum allowed time, in milliseconds, between poll cycles while processing messages. Increase this value for long-running workflows.                              |
| topicMetadataRefreshIntervalMs | Integer  | 1 – 3,600,000                                                     | 300000                   | Interval, in milliseconds, for refreshing Kafka topic metadata. Controls how quickly newly created topics matching a regex event subscription are discovered. Reduce it for shorter discovery delay, at the cost of additional metadata traffic. |
| producerAcks                   | Enum     | None<br>Leader<br>All                                             | All                      | Kafka producer acknowledgement mode. None does not wait for broker acknowledgement, Leader waits for the partition leader, and All waits for all in-sync replicas. |
| compressionType                | Enum     | None<br>Gzip<br>Snappy<br>Lz4<br>Zstd                             | None                     | Default Kafka message compression for producer sends. None sends messages without compression.                                                                     |
| connectionLossFailureThreshold | Integer  |                                                                   | 3                        | Number of consecutive low-level Kafka connectivity failures required to mark the driver as disconnected. Set to 0 to disable this detector.                        |
| checkTopicsOnConnect           | Boolean  |                                                                   | false                    | Validate configured literal event topics against broker metadata during setup before consumption starts.                                                           |
| maxAutoCommitMessagesPerSecond | Integer  |                                                                   | 10                       | Maximum number of automatic offset commits per second, across all consumed topics and partitions. Set it to 0 to disable rate limiting. It applies only to events with autoCommit enabled.                        |

Table: Kafka parameters

## Extended Parameters

### Event

| Name          | Type    | Possible Values  | Default Value | Description                                                                                                                  |
| ------------- | ------- | ---------------- | ------------- | ---------------------------------------------------------------------------------------------------------------------------- |
| topicType     | Enum    | Literal<br>Regex | Literal       | How the topic value is matched. Literal performs a direct string comparison, and Regex performs regular expression matching. |
| autoCommit    | Boolean |                  | true          | Commit the consumed offset immediately after the driver emits the event.                                                     |

Table: Kafka event extended parameters

!!! note
    Reading from the beginning of a topic is controlled by the driver-level `fromBeginning` protocol parameter, not per event — see [Protocol Parameters](#protocol-parameters).

### Event Property

- _Not Applicable_

### Property

| Name         | Type | Possible Values                                                                                                     | Default Value | Description                                                                                                                          |
| ------------ | ---- | ------------------------------------------------------------------------------------------------------------------- | ------------- | ------------------------------------------------------------------------------------------------------------------------------------ |
| propertyType | Enum | Topic<br>Partition<br>Offset<br>Key<br>Headers<br>HeadersJson<br>Timestamp<br>Value<br>ValueBase64<br>ConsumerGroup |               | Kafka message field represented by the property. This value is required. The driver normalizes the Device Property Id to this value. |

Table: Kafka specific extended properties

## Command

The protocol supports the following commands, selected through the `commandType` attribute:

| Name            | Type | Possible Values                                                                                  | Default Value    | Description                                                                                                    |
| --------------- | ---- | ------------------------------------------------------------------------------------------------ | ---------------- | -------------------------------------------------------------------------------------------------------------- |
| commandType     | Enum | Publish<br>Commit<br>SendToDLQ<br>PauseTopic<br>ResumeTopic<br>PauseAllTopics<br>ResumeAllTopics | Publish          | Kafka command to run. SendToDLQ publishes to the configured Topic command parameter.                           |
| producerAcks    | Enum | UseGlobalSetting<br>None<br>Leader<br>All                                                        | UseGlobalSetting | Producer acknowledgement override for this command. UseGlobalSetting uses the driver-level producerAcks value. |
| compressionType | Enum | UseGlobalSetting<br>None<br>Gzip<br>Snappy<br>Lz4<br>Zstd                                        | UseGlobalSetting | Compression override for this command. UseGlobalSetting uses the driver-level compressionType value.           |

Table: Kafka commands

The available commands are:

- **Publish** - sends a message to a Kafka topic.
- **Commit** - commits an offset for a topic and partition. Requires the partition and offset parameters.
- **SendToDLQ** - publishes to the dead-letter queue. This command uses the publish pipeline and sends to the configured Topic command parameter.
- **PauseTopic** - pauses Kafka consumption for a topic and all of its partitions. Requires the topic parameter.
- **ResumeTopic** - resumes Kafka consumption for a topic and all of its partitions. Requires the topic parameter.
- **PauseAllTopics** - pauses Kafka consumption for all subscribed topics.
- **ResumeAllTopics** - resumes Kafka consumption for all subscribed topics.

## Command Parameters

Command parameters are classified by the `parameterCategory` attribute:

| Category   | Description                                                                                                                                                                                                               |
| ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Topic      | Kafka topic used by Publish, SendToDLQ, Commit, PauseTopic, and ResumeTopic. Exactly one topic parameter is required for these commands.                                                                                  |
| Key        | Kafka message key. This parameter is optional, with at most one per command.                                                                                                                                              |
| HeaderBase | Object or JSON object string used as the base headers object. This parameter is optional, with at most one per command.                                                                                                   |
| Header     | Individual header merged into the base headers. The header name is the command parameter name, and the header value is the runtime parameter value.                                                                       |
| ValueBase  | Whole payload value. This is the default command parameter category and, when no Value parameters are configured, is published as the raw full message payload. This parameter is optional, with at most one per command. |
| Value      | Individual value field merged into the base payload object. The field name is the command parameter name, and the field value is the runtime parameter value.                                                             |
| Offset     | Offset-related command parameter. Commit requires the partition and offset parameters.                                                                                                                                    |
| Option     | Command option. A runtime compressionType override is applied only when the parameter is categorized as Option and named compressionType.                                                                                 |

Table: Kafka command parameter categories

## Event Properties

When a Kafka message is consumed and matched to a configured event, the following properties are emitted in the event occurrence:

| Name          | Type     | Description                                                                      |
| ------------- | -------- | -------------------------------------------------------------------------------- |
| Topic         | String   | The Kafka topic the message was consumed from.                                   |
| Partition     | Integer  | The partition number.                                                            |
| Offset        | String   | The message offset within the partition.                                         |
| Key           | String   | The message key, UTF-8 decoded. Returns an empty string if null.                 |
| Headers       | Object   | The message headers as a key-value object. Multi-value headers are comma-joined. |
| HeadersJson   | String   | The message headers serialized as a JSON string.                                 |
| Timestamp     | DateTime | The message timestamp. Falls back to the current date if not set by the broker.  |
| Value         | String   | The message payload, UTF-8 decoded. Returns an empty string if null.             |
| ValueBase64   | String   | The message payload as a Base64-encoded string.                                  |
| ConsumerGroup | String   | The consumer group ID used to consume the message.                               |

Table: Kafka event properties

## Remarks/Behavior

### Consumer lifecycle

- The Kafka consumer is created only when at least one enabled event with a non-empty Event Device Id topic is configured. If no enabled events are configured, the driver runs in producer-only mode and does not create a consumer connection to the broker.
- Topic subscriptions are derived from enabled events during setup completion.
- The driver subscribes and immediately applies the startup pause before entering the **Communicating** state, then starts or resumes consumption only after **Communicating**. This guarantees that no Kafka message is consumed before setup is fully completed.

### Message routing

When a Kafka message is received, the driver routes it to configured events whose Event Device Id topic or pattern matches the incoming message topic. The Event Device Id is mandatory and unique.

- Literal matching - the event's Device Event Id must exactly equal the message topic.
- Regex matching - the event's Device Event Id is compiled as a JavaScript regular expression and tested against the message topic.
- When an event is emitted, the `eventDeviceId` uses the configured event device id, which is the Kafka topic or regex pattern.
- The original Kafka topic is exposed through an event property configured with a `propertyType` of `Topic`, including regex matches.
- Event property values are produced from the event's configured properties using each property's `propertyType`.
- Different regex patterns can match the same incoming topic. Each matching event emits an independent occurrence.

!!! note
    Topics created after a regex subscription is established are not matched immediately — they are discovered on the next topic metadata refresh, within `topicMetadataRefreshIntervalMs`. Reduce this value for shorter discovery delay, at the cost of additional metadata traffic — see [Protocol Parameters](#protocol-parameters).

### Offset management

The driver supports two offset commit strategies, configured per event through extended data:

- Auto commit (default) - the driver commits the offset immediately after emitting the event occurrence. This is checked per matching event. If any matching event has `autoCommit` enabled, the offset is committed after all event occurrences are emitted.
- Manual commit - set the event `autoCommit` extended data to false and commit explicitly through the Commit command with the topic, partition, and offset parameters. The committed offset is offset+1, which is the next message to read.

!!! warning
    When multiple events match the same Kafka message, a single matching event with autoCommit enabled is enough to commit the offset. Avoid overlapping an event that must be manually committed with a catch-all or regex event that has auto commit enabled.

Use the `maxAutoCommitMessagesPerSecond` protocol parameter to cap how many automatic offset commits the driver performs per second, across all consumed topics and partitions. When the cap is reached, the driver waits for capacity before committing, so no offset is discarded. Set the parameter to 0 to disable rate limiting.

!!! note
    Rate limiting applies only to automatic commits. Offsets committed through the Commit command are never rate-limited.

### Connection loss detection

The driver watches low-level Kafka connectivity errors from librdkafka. After `connectionLossFailureThreshold` consecutive failures, within an internal window, the driver transitions to the **Disconnected** state, triggering the base driver reconnect mechanism. Set `connectionLossFailureThreshold` to 0 to disable this detector.

### Pause and resume

The PauseTopic and ResumeTopic commands pause or resume Kafka consumption for a topic and all of its partitions. The PauseAllTopics and ResumeAllTopics commands act on all subscribed topics. The pause state is preserved across reconnects.

### Topic validation

When `checkTopicsOnConnect` is enabled, the driver validates configured literal event topics against broker metadata during setup, before consumption starts.

### Logging

Kafka payloads, headers, and low-level client diagnostics can be large. Configure `logging[].options.maxLength` in the Connect IoT `config.json` for the file transports used by controllers and drivers. This setting limits the size of each log message and keeps the beginning and end of oversized entries with an omitted-character marker in the middle, reducing the risk of writing very large message dumps to disk.

Example:

```json
{
    "id": "ControllersAndDrivers",
    "type": "File",
    "options": {
        "level": "debug",
        "maxLength": 1024
    },
    "applications": ["AutomationController", "Driver*"]
}
```

Low-level Kafka client logs are prefixed with `##`. Use the `logLowLevelMessages` protocol parameter to control the minimum level routed through the driver logger, or set it to `None` to disable these logs.

### Example: catch-all event

An event can be configured to expose every Kafka field out of the box by using a catch-all regex as the Event Device Id, and then associating all Kafka properties with that event.

Example event configuration:

```json
{
    "Name": "Kafka AllEvents Received",
    "DeviceEventId": "^.*",
    "ExtendedData": {
        "topicType": "Regex",
        "autoCommit": true
    }
}
```

Example event properties for that event:

| Name          | Data Type | propertyType  |
| ------------- | --------- | ------------- |
| Topic         | String    | Topic         |
| Partition     | Integer   | Partition     |
| Offset        | String    | Offset        |
| Key           | String    | Key           |
| Headers       | Object    | Headers       |
| HeadersJson   | String    | HeadersJson   |
| Timestamp     | DateTime  | Timestamp     |
| Value         | String    | Value         |
| ValueBase64   | String    | ValueBase64   |
| ConsumerGroup | String    | ConsumerGroup |

Table: Catch-all event properties

In the GUI, the event identity remains the configured regex pattern, where the Device Event Id is `^.*`, while the original consumed topic is available by adding a property whose `propertyType` is `Topic`.