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 ↗ 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:

Table: Kafka specific data types

Protocol Parameters#

The protocol supports the following parameters:

Table: Kafka parameters

Extended Parameters#

Event#

Table: Kafka event extended parameters

Event Property#

• Not Applicable

Property#

Table: Kafka specific extended properties

Command#

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

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:

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:

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.

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.

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.

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:

{
    "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:

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

Example event properties for that event:

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.