> ## Documentation Index > Fetch the complete documentation index at: https://docs.automq.com/llms.txt > Use this file to discover all available pages before exploring further. # Azure IoT Hub Sink Connector > Configure and operate the Azure IoT Hub Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## 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](../manage-connectors). The following configuration reads JSON strings from a topic and sends each message to the device specified by its `deviceId`. ```properties theme={null} connector.class=com.microsoft.azure.iot.kafka.connect.sink.IotHubSinkConnector topics= value.converter=org.apache.kafka.connect.storage.StringConverter IotHub.ConnectionString= ``` Replace `` with the input topic and `` 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: ```json theme={null} {"messageId":"msg-1001","message":"reboot","deviceId":"device-001"} ``` 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**: ```properties theme={null} connector.class=com.microsoft.azure.iot.kafka.connect.sink.IotHubSinkConnector topics.regex=devices\..*\.commands value.converter=org.apache.kafka.connect.storage.StringConverter IotHub.ConnectionString= ``` Replace `` 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**: ```properties theme={null} connector.class=com.microsoft.azure.iot.kafka.connect.sink.IotHubSinkConnector topics= tasks.max=2 value.converter=org.apache.kafka.connect.storage.StringConverter IotHub.ConnectionString= ``` Replace `` and `` 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](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), 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.