Skip to main content

​
Overview

The Azure IoT Hub Sink Connector reads cloud-to-device messages from a Kafka topic, converts each record into an Azure IoT Hub message, and sends it to the device specified by the record. It sits between the Kafka event stream and the IoT Hub device messaging channel, and is suitable for forwarding device commands, notifications, or configuration updates generated by business systems to IoT Hub. The record value can be represented by a Struct or a JSON string. A message must contain at least messageId, message, and deviceId; deviceId selects the target device. expiry can be used in a Struct to specify the message expiration time, while the corresponding property name in a JSON string is expiryTime. The Connector does not handle device feedback, direct methods, or other cloud-to-device communication patterns.

​
Prerequisites

  • The target device that receives messages must exist in Azure IoT Hub, and the connection string must have permission to send cloud-to-device messages to IoT Hub. If message expiration is used, also confirm that the business rules of the target Azure IoT Hub and device side match the expected behavior.

​
License

Uses the MIT License.

​
Quick Start

Prepare the Connect Cluster, Kafka, and Azure IoT Hub in advance, confirm that the Worker can access the input topic and Azure IoT Hub, and prepare the target device. For cluster and Connector management operations, see Manage Connectors. The following configuration reads JSON strings from a topic and sends each message to the device specified by its deviceId.
Replace <input-topic> with the input topic and <iot-hub-connection-string> with the Azure IoT Hub connection string. Do not write the connection string to logs or commit it to version control. When using this example, the string value of each message in the topic should be parseable JSON, for example:
The optional expiryTime in JSON must be a parseable ISO-8601 Instant; omitting this property means that no expiration time is set. Record conversion fails when an input record is missing a required field, a field type does not match, or the JSON cannot be parsed.

​
Configuration

​
Connection and Messages

​
IotHub.ConnectionString

Sets the Azure IoT Hub connection string.
  • Type: string
  • Default: None
  • Importance: High
  • Required: Yes
  • Valid Values / Notes: Must provide a connection string that can be used to create an Azure IoT Hub ServiceClient. The configuration definition does not validate whether the value is non-empty, correctly formatted, or reachable; an invalid value may cause failure when the Connector starts or sends a message. This configuration contains sensitive credentials. Use a secure configuration injection method and do not output it to logs.

​
IotHub.MessageDeliveryAcknowledgement

Sets the message delivery acknowledgement mode for messages written to Azure IoT Hub.
  • Type: string
  • Default: None
  • Importance: High
  • Valid Values / Notes: Only the case-sensitive values None, Full, PositiveOnly, or NegativeOnly can be used. The value is sent as acknowledgement metadata for each Azure message; it does not prove that the device has received, processed, or returned feedback for the message, and it does not generate a Kafka acknowledgement record.

​
Input Subscription and Tasks

​
connector.class

Selects the Azure IoT Hub Sink Connector implementation class.
  • Type: string
  • Default: None
  • Importance: High
  • Required: Yes
  • Valid Values / Notes: Use com.microsoft.azure.iot.kafka.connect.sink.IotHubSinkConnector.

​
topics

Specifies the list of Kafka topics to read.
  • Type: list
  • Default: Empty list []
  • Importance: High
  • Required: Mutually exclusive with topics.regex
  • Valid Values / Notes: Use comma-separated topic names and configure a non-empty topics instead of topics.regex; setting both or leaving both empty causes Sink configuration validation to fail.

​
topics.regex

Subscribes to Kafka topics using a regular expression.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Required: Mutually exclusive with topics
  • Valid Values / Notes: Use a non-empty Java regular expression instead of topics; setting both or leaving both empty causes Sink configuration validation to fail.

​
tasks.max

Sets the maximum number of Tasks that the Connector can create.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid Values / Notes: Must be at least 1. Actual parallelism is limited by the number of Kafka partitions and the assignment result; increasing this value does not guarantee per-device serialization and does not establish global ordering across partitions or Tasks.

​
Data Conversion

​
key.converter

Converts Kafka record keys into Kafka Connect data.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid Values / Notes: When not set, the Worker-level key converter is used. When set, it must be an instantiable org.apache.kafka.connect.storage.Converter implementation. The Connector uses the record value to select the device and does not read the record key to map the target device.

​
value.converter

Converts Kafka record values into Kafka Connect data for the Connector to map to Azure IoT Hub messages.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid Values / Notes: When not set, the Worker-level value converter is used. When set, it must be an instantiable org.apache.kafka.connect.storage.Converter implementation. The converted value must conform to a String or Struct path supported by the Connector. A String value should be JSON containing messageId, message, deviceId, and optional expiryTime; a Struct value must contain same-named string fields messageId, message, and deviceId, and may contain a string-typed expiry field.

​
Best Practices

​
Subscribe to a Group of Input Topics by Regular Expression

Applicable business scenario: You are connecting multiple message topics with consistent naming rules for the first time, or will continue creating topics of the same type and want the Connector to automatically include matching topics without modifying a fixed topic list each time. Configuration example:
Replace <iot-hub-connection-string> with the Azure IoT Hub connection string. After using topics.regex, remove topics; do not configure both subscription methods. String values in each topic matched by the expression should follow the same JSON message structure. Key point: Regular-expression subscription makes it easier to manage topics of the same type uniformly, but the topic naming rules directly determine the consumption scope. Before changing the naming rules, confirm whether the old and new topics will match at the same time to avoid unintentionally expanding or shrinking the input scope.

​
Increase Task Parallelism by Partition

Applicable business scenario: The Connector is already sending messages reliably, the input topic has multiple partitions, and the per-record synchronous sending throughput of a single Task is insufficient. You want to process partitions in parallel by increasing the number of Tasks. Configuration example:
Replace <input-topic> and <iot-hub-connection-string> with the actual values. First confirm that the input topic has at least two assignable partitions, then increase tasks.max gradually while observing Task status, sending latency, and consumer Lag. Key point: tasks.max is an upper limit, not the actual number of parallel Tasks; Kafka Connect creates and assigns Tasks based on the partition assignment result. Each Task sends messages independently. The Connector does not guarantee processing order for the same device across partitions or Tasks, so when ordering is required, an appropriate partition key should be designed upstream.

​
Monitoring

​
What to Monitor

Monitor the health status and status changes of the Kafka Connect Worker, Connector, and Tasks, and pay attention to input throughput, sending latency, consumer Lag, Offset commits, errors, and retries. Also observe CPU, memory, and garbage-collection signals for the Worker JVM. Because the Connector sends each record synchronously, increased sending latency may slow Task processing. If Kafka Connect error handling or a dead-letter queue is enabled, also monitor the corresponding error-handling counters and DLQ activity. Do not treat IotHub.MessageDeliveryAcknowledgement as a device feedback metric.

​
Import the Grafana Dashboard

Obtain the Dashboard JSON from Download the Connect Cluster Grafana Dashboard, prepare a Prometheus data source that collects Kafka Connect metrics, and confirm that the label names match the data source configuration. Then import the JSON into Grafana and select the corresponding data source.

​
Limitations

  • The Connector only handles cloud-to-device messages; it does not provide device feedback, consumption of acknowledgement results, direct methods, or other Azure IoT Hub communication patterns.
  • Each input record is converted and sent synchronously and individually. There are no Connector-level batch requests, asynchronous sending, rate limiting, or configurable in-flight message count.
  • Exactly one of topics and topics.regex must be configured with a non-empty value; they cannot be used together or both omitted.
  • Successful message sending and Kafka Offset commits are not one atomic operation. If the process fails after sending completes but before the Offset is committed, the record may be sent again after recovery.
  • The Connector does not provide global ordering across partitions, Tasks, or by device, and does not provide idempotent deduplication or exactly-once delivery guarantees.
  • Struct input requires messageId, message, and deviceId to be string fields. String input must be JSON that can be deserialized into a message object; incorrect fields or formats cause record conversion to fail.

​
FAQ

​
What should I do if the Connector reports a topics and topics.regex configuration error when starting?

Check whether the two configurations are both empty or both set. For a fixed-topic scenario, keep only a non-empty topics; to subscribe by naming rules, keep only a non-empty topics.regex. Submit the Connector configuration again after making the change, and confirm that the regular expression is a valid Java regular expression.

​
How can I check input records when message conversion fails?

First check the conversion error in the Task log, then confirm that the value converter outputs a String or Struct. A String value should be JSON containing string-typed messageId, message, and deviceId; a Struct value should contain these fields and use a string-typed expiry for the optional expiration time. The expiration field name in JSON is expiryTime; do not confuse it with the Struct field name expiry.

​
Why might a Kafka record be sent again after a successful send call?

Connector sending and Kafka Offset committing are two separate steps. If the process fails after the Azure service accepts the send request but before the Offset is committed, the record may be read and sent again after recovery. Check the Task status, Offset commit status, and consumer Lag, and handle potential duplicates according to the message’s business idempotency design. Do not interpret the acknowledgement mode as a cross-system exactly-once guarantee.

​
Is it normal for processing order to change after increasing tasks.max?

Yes. After the number of Tasks increases, Kafka may assign different partitions to different Tasks. The Connector only preserves the iteration order of records received in a single processing call by an individual Task; it does not guarantee global ordering across partitions, Tasks, or by device. When ordering is required, put related messages in the same partition and control parallelism.