> ## 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. # Ably Sink Connector > Configure and operate the Ably Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## Overview The Ably Sink Connector publishes events from Kafka topics to Ably channels, connecting backend event streams to real-time subscribers. It supports use cases such as status updates, business event notifications, and live dashboards. Each Kafka record maps to one Ably message. The destination channel can be fixed or selected dynamically from the topic, partition, or record fields; message names can distinguish event types. Message content depends on the output of the Kafka Connect converter. Schema-backed Struct values are converted to JSON without the schema definition; strings and bytes follow their respective message conversion paths. Schemaless Map or List values are not automatically converted to structured JSON. To publish JSON content, provide a JSON string or a schema-backed Struct. Consumption resumes from offsets committed by Kafka Connect. Duplicates can occur if a failure happens after publishing succeeds but before offsets are committed, or if a request is retried after its response is lost. Do not treat this connector as an exactly-once or unconditional at-least-once delivery solution. For business events that must not be lost or processed twice, include a business event ID in the payload and implement downstream deduplication, reconciliation, and redelivery. ## Prerequisites * The Ably API key must have `publish` permission for every static or dynamic destination channel. * Nested field templates require a converter that provides the corresponding Connect Schema and Struct. ## License Licensed under Apache License 2.0. ## Quick Start Prepare a Connect cluster, Kafka topics, an Ably account, and destination channels, and verify network connectivity and access permissions. For cluster preparation and connector management, see [Manage Connectors](../manage-connectors). The following example consumes UTF-8 string messages and publishes them to a fixed channel. ```properties theme={null} connector.class=com.ably.kafka.connect.ChannelSinkConnector topics= channel= client.key= client.id= value.converter=org.apache.kafka.connect.storage.StringConverter ``` Replace the topic, channel, API key, and client identity, then apply the example as the connector configuration. `client.id` is a nonempty Ably client identity, not a Kafka consumer client identifier. The example does not use record keys, so it inherits the worker's key converter. To use key templates, also verify how keys are decoded. Inject credentials through your deployment environment's secure configuration mechanism; do not store them in version control or shared logs. ## Configuration The following reference lists all public plugin settings and six Kafka Connect framework settings that directly affect subscriptions, parallelism, and message decoding. A default of None means there is no default value; required settings must be supplied explicitly. `null` means unspecified, and `[]` means an empty list; neither is equivalent to an empty string. Configuration names are case-sensitive. ### Connector and Subscriptions #### `connector.class` Specifies the sink connector implementation class. * **Type**: `STRING` * **Default**: None * **Required**: Yes * **Importance**: High * **Valid Values / Notes**: Use `com.ably.kafka.connect.ChannelSinkConnector`. Every worker running a task must be able to load the plugin. #### `topics` Specifies the Kafka topics to consume. * **Type**: `LIST` * **Default**: `[]` * **Importance**: High * **Valid Values / Notes**: Separate multiple names with commas. Configure exactly one of `topics` and `topics.regex`. When a DLQ is enabled, do not subscribe to the DLQ topic. #### `topics.regex` Selects Kafka topics to consume using a regular expression. * **Type**: `STRING` * **Default**: `""` * **Importance**: High * **Valid Values / Notes**: Uses Java regular expression syntax. Mutually exclusive with `topics`; when a DLQ is enabled, the expression must not match the DLQ topic. #### `tasks.max` Sets the maximum number of tasks. * **Type**: `INT` * **Default**: `1` * **Importance**: High * **Valid Values / Notes**: Must be at least `1`. Actual consumption parallelism is limited by the number of subscribed partitions. A partition is not assigned simultaneously to multiple tasks in the same consumer group. This setting controls a different dimension of parallelism from the batch execution thread count within each task. ### Authentication and Logging #### `client.key` Sets the Ably API key used to publish messages. * **Type**: `PASSWORD` * **Default**: None * **Required**: Yes * **Importance**: High * **Valid Values / Notes**: Use a valid API key with permissions covering all destination channels. Do not expose the actual value in configuration examples, version control, or logs. #### `client.id` Sets the Ably client identity. * **Type**: `STRING` * **Default**: None * **Required**: Yes * **Importance**: High * **Valid Values / Notes**: A nonempty string; this is not the Kafka consumer's `client.id`. #### `client.loglevel` Sets the Ably SDK logging level. * **Type**: `INT` * **Default**: `2` * **Importance**: Low * **Valid Values / Notes**: Defined levels are `2` (VERBOSE), `3` (DEBUG), `4` (INFO), `5` (WARN), `6` (ERROR), and `99` (NONE). In production, choose a level appropriate for troubleshooting and avoid excessive debug output over extended periods. ### Message Decoding and Mapping #### `key.converter` Sets the converter for Kafka record keys. * **Type**: `CLASS` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: Inherits the worker setting when unspecified. When using `#{key}` or `#{key.}`, choose a converter that matches the actual encoding. Nested paths require Schema/Struct. #### `value.converter` Sets the converter for Kafka record values. * **Type**: `CLASS` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: Inherits the worker setting when unspecified. Strings can use `org.apache.kafka.connect.storage.StringConverter`. `#{value.}` requires Schema/Struct, not just JSON text or a schemaless Map. Configure converter-specific options according to the selected converter. #### `channel` Sets a static name or template for the destination Ably channel. * **Type**: `STRING` * **Default**: None * **Required**: Yes * **Importance**: High * **Valid Values / Notes**: A nonempty string whose first character is not a colon, comma, whitespace, or `[`. The name must not contain a newline. Templates support `#{topic}`, `#{topic.name}`, `#{topic.partition}`, `#{key}`, `#{key.}`, and `#{value.}`. Nested paths traverse Struct fields; array indexing and Map paths are not supported. Mapping errors such as missing fields are controlled by `onFailedRecordMapping`. Template output must also satisfy Ably channel naming and permission requirements. #### `message.name` Sets the Ably message name to identify the event type. * **Type**: `STRING` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: Accepts static text or the same templates as `channel`. No name is set when unspecified or empty. There is no direct `#{value}` placeholder. #### `messagePayloadSizeMax` A public setting related to message payload size. * **Type**: `INT` * **Default**: `65536` * **Importance**: Medium * **Valid Values / Notes**: This setting does not enforce a payload size limit. Do not rely on it to validate, truncate, or split messages. Control actual message and request sizes upstream and comply with Ably service limits. ### Batching and Concurrency #### `batchExecutionThreadPoolSize` Sets the number of batch publishing threads per task. * **Type**: `INT` * **Default**: `10` * **Importance**: Medium * **Valid Values / Notes**: Must be greater than `0`. Set to `1` to execute batch publishing serially within the task. Multiple threads allow batches to complete concurrently, without guaranteeing completion order across batches or global ordering across tasks. #### `batchExecutionMaxBufferSize` Sets the maximum number of Kafka records buffered before submitting a batch. * **Type**: `INT` * **Default**: `100` * **Importance**: Medium * **Valid Values / Notes**: Use a value greater than `0`. A batch is submitted when the record count reaches this threshold. This is neither a byte-size limit nor a limit on the total backlog of data awaiting publication. #### `batchExecutionMaxBufferSizeMs` Sets the wait time in milliseconds for timed publishing of a partially filled buffer. * **Type**: `INT` * **Default**: `100` * **Importance**: Medium * **Valid Values / Notes**: Use a nonnegative value; `0` can trigger timed publishing immediately. Full batches can be published earlier. This is neither a minimum per-record delay nor an upper bound on end-to-end latency. #### `client.async.http.threadpool.size` Sets the Ably SDK asynchronous HTTP thread pool size. * **Type**: `INT` * **Default**: `64` * **Importance**: Medium * **Valid Values / Notes**: This controls SDK asynchronous requests, not this connector's batch publishing concurrency. Adjust batch publishing concurrency with `batchExecutionThreadPoolSize`. ### Connections and Fallback Endpoints #### `client.tls` Controls whether connections to Ably use TLS. * **Type**: `BOOLEAN` * **Default**: `true` * **Importance**: Medium * **Valid Values / Notes**: `true` or `false`. Keep enabled in production to avoid transmitting credentials and messages in plaintext. #### `client.rest.host` Sets a custom REST host. * **Type**: `STRING` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: The SDK selects the REST host when unspecified. Set only when a custom endpoint is needed, and verify that authentication, networking, and fallback endpoint settings are consistent. When using a nondefault REST host, do not also set `client.environment`. #### `client.port` Sets the port for non-TLS connections. * **Type**: `INT` * **Default**: `0` * **Importance**: Low * **Valid Values / Notes**: `0` uses the SDK's default non-TLS port, `80`. Used only when `client.tls=false`. An explicit value must be a valid service port. #### `client.tls.port` Sets the port for TLS connections. * **Type**: `INT` * **Default**: `0` * **Importance**: Low * **Valid Values / Notes**: `0` uses the SDK's default TLS port, `443`. Used only when `client.tls=true`. An explicit value must be a valid service port. #### `client.environment` Sets a nondefault Ably service environment. * **Type**: `STRING` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: Configure only when required by the Ably service deployment; endpoint selection is handled by the SDK. It cannot be combined with a nondefault `client.rest.host`. This is not a Kafka Connect or managed platform environment identifier. #### `client.http.open.timeout` Sets the timeout in milliseconds for establishing an HTTP connection. * **Type**: `INT` * **Default**: `4000` * **Importance**: Medium * **Valid Values / Notes**: Set a reasonable nonnegative timeout for your network conditions. Do not treat it as a timeout for the entire message processing workflow. #### `client.http.request.timeout` Sets the timeout in milliseconds for a single HTTP request and response. * **Type**: `INT` * **Default**: `10000` * **Importance**: Medium * **Valid Values / Notes**: Set a reasonable nonnegative timeout for network conditions and request size. It does not include the entire Kafka Connect error retry window. #### `client.http.max.retry.count` Sets the number of SDK HTTP retries against fallback hosts. * **Type**: `INT` * **Default**: `3` * **Importance**: Medium * **Valid Values / Notes**: A nonnegative value is recommended. Retries require available fallback hosts and an error that meets the SDK's host-failure criteria. This setting does not retry every HTTP error and is not a Kafka consumption retry policy. #### `client.fallback.hosts` Sets the list of fallback hosts available for HTTP requests. * **Type**: `LIST` * **Default**: `[]` * **Importance**: Medium * **Valid Values / Notes**: Separate multiple hosts with commas; they must match the actual Ably environment. An empty list provides no fallback candidates and does not automatically enable the SDK's built-in fallback hosts. Increasing the retry count alone does not enable fallback-host retries. ### Proxy #### `client.proxy` Controls whether an HTTP proxy is enabled. * **Type**: `BOOLEAN` * **Default**: `false` * **Importance**: Medium * **Valid Values / Notes**: When enabled, a proxy host and actual port are required. Proxy parameter types and enum values must remain valid even when the proxy is disabled. #### `client.proxy.host` Sets the HTTP proxy host. * **Type**: `STRING` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: When `client.proxy=true`, specify a nonempty, reachable proxy host. #### `client.proxy.port` Sets the HTTP proxy port. * **Type**: `INT` * **Default**: `0` * **Importance**: Medium * **Valid Values / Notes**: When the proxy is enabled, specify an actual valid port. The default `0` does not automatically select `80` or `443`. #### `client.proxy.username` Sets the proxy authentication username. * **Type**: `STRING` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: This is not an Ably identity. When the proxy is enabled and a username is set, `client.proxy.password` must also be set. #### `client.proxy.password` Sets the proxy authentication password. * **Type**: `PASSWORD` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: Configure together with the proxy username. Inject through a secure configuration mechanism and do not log the actual value. #### `client.proxy.pref.auth.type` Sets the preferred proxy authentication type. * **Type**: `STRING` * **Default**: `BASIC` * **Importance**: Medium * **Valid Values / Notes**: Only `BASIC`, `DIGEST`, and `X_ABLY_TOKEN` are allowed; values are case-sensitive. Choose a type supported by the proxy. #### `client.proxy.non.proxy.hosts` Sets the list of hosts that bypass the proxy. * **Type**: `LIST` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: Separate multiple hosts with commas. The SDK's proxy configuration handles this list; do not assume arbitrary wildcard syntax is supported. ### Error Handling and Push Options #### `onFailedRecordMapping` Sets the handling policy for channel or message name mapping failures. * **Type**: `STRING` * **Default**: `stop` * **Importance**: Medium * **Valid Values / Notes**: Only `stop`, `skip`, and `dlq` are allowed; values are case-sensitive. `stop` stops batch processing, while `skip` discards records that fail mapping. `dlq` requires an available Kafka Connect error record reporting mechanism, a configured `errors.deadletterqueue.topic.name`, and `errors.tolerance=all` to allow continued processing. DLQ-mode processing cannot continue without the reporting mechanism. This setting does not cover all converter, content conversion, or HTTP publishing errors. #### `client.push.full.wait` Sets the wait option passed to the Ably SDK for Push REST operations. * **Type**: `BOOLEAN` * **Default**: `false` * **Importance**: Medium * **Valid Values / Notes**: This sink does not perform device Push management operations. The setting does not control batch publishing completion or guarantee delivery of device notifications. Notification use cases also require separate Ably Push configuration and device registration. ## Best Practices ### Reduce Batch Interleaving for Status Events **Applicable Scenario**: Order or device status events are already partitioned by business entity, and you want to reduce interleaving caused by concurrently published batches within the same task, rather than achieve global ordering across entities. **Configuration Example**: Add the following setting to the Quick Start configuration to override the default batch execution thread count. ```properties theme={null} batchExecutionThreadPoolSize=1 ``` **Key Considerations**: This setting makes batch publishing serial within one task, but multiple tasks and partitions can still publish concurrently to the same channel. It does not guarantee strict end-to-end ordering or duplicate-free delivery. Write events for the same entity to the same Kafka partition, include an entity version or business sequence number in the payload, and let subscribers identify duplicates and stale states. Also evaluate the throughput impact of serial publishing. Do not rely solely on increasing the buffered record count to address a persistent backlog; use consumer lag, request duration, and worker memory to guide adjustments to the production rate and publishing parallelism. ## Monitoring ### What to Monitor Monitor Kafka Connect cluster and worker health, connector and task status, throughput, consumption and processing latency, offset commits and commit failures, errors, and retries. Use worker JVM heap usage, GC, and thread activity to assess backlogs. A running task or a committed offset alone does not prove that business clients have received every message; also reconcile business events. Monitor DLQ activity only when the corresponding error handling is enabled. ### Import the Grafana Dashboard Download the [Kafka Connect Grafana Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json). Ensure that a Prometheus-compatible data source collects Kafka Connect metrics and that labels such as cluster and worker match the dashboard queries, then import the JSON into Grafana and select the corresponding data source. ## Limitations * Exactly-once delivery from Kafka to Ably is not guaranteed. Do not rely on automatic idempotent publishing to eliminate duplicates. * Global ordering across partitions, tasks, or channels is not provided. * Schema-backed values support only top-level Struct, String, and Bytes, not top-level numeric, Boolean, Array, or Map values. * Nested templates require Schema/Struct and do not support parsing JSON strings, schemaless Maps, array indexing, or Map paths. * Template leaf values support strings, Integer, Long, Boolean, and byte arrays containing valid UTF-8, but not Float, Double, Byte, Short, or ByteBuffer. * Struct-to-JSON conversion does not support direct raw byte-array fields. Do not assume that arbitrary logical types or timestamp precision are preserved unchanged. * Without a value schema, Kafka keys, headers, and Push Extras are not exported. Even with a schema, the topic, partition, offset, and Kafka timestamp are not automatically exported. * A null value does not delete Ably data, and Kafka tombstones are not automatically skipped. * The buffered record count does not limit the total pending backlog; persistently slow requests can cause memory pressure. * `messagePayloadSizeMax` does not limit actual payload size. Message sizes, batch request sizes, and rates remain subject to Ably service limits. ## FAQ ### Why Are There No Messages in the Destination Channel While the Task Is Running? Check whether the topic has new records, whether the consumer group is already at the end, whether the subscription matches, and whether the channel produced by the template matches the subscriber's channel. Verify that the API key has publish permission for that channel, then check REST errors, connection timeouts, and worker memory. At low traffic levels, also account for the timed publishing wait for partially filled batches. A running status alone does not indicate successful publishing. ### Why Does a Field Template Fail Even Though the Field Exists in the JSON? JSON strings and schemaless Maps cannot supply data for nested templates. Check whether the converter outputs a schema-backed Struct, whether the field path exists, and whether the leaf value has a supported type. Adjust the converter or upstream event structure to match the actual Kafka encoding. Field renames and removals also affect templates, so check referenced fields before deploying schema changes. ### Why Is the Message Not the Expected JSON Object, or Why Are Keys and Headers Missing? Schemaless Map/List values are not automatically published as structured JSON, and Key and Headers Extras are not generated without a value schema. For JSON content, use a JSON string or a schema-backed Struct. For Extras, use a supported schema-backed value and check how keys and headers are decoded. String keys remain strings, byte-array keys use Base64, and other key types are not automatically exported. ### Why Is Publishing Not Retried After a Failure Even Though I Increased the Retry Count? Check whether `client.fallback.hosts` specifies available fallback hosts matching the environment. The default empty list provides no fallback candidates. The SDK retries against fallback hosts only for errors that meet its host-failure criteria; do not expect permission errors, rate limiting, or all server errors to be retried automatically. Address credentials, permissions, or rate issues first. Increasing the retry count is not a substitute for these fixes. ### Why Are Records Not Sent to the DLQ After Mapping Fails? Verify that `onFailedRecordMapping=dlq` and check the Kafka Connect DLQ topic, write permissions, error tolerance settings, and availability of the error record reporting mechanism. Setting the plugin policy alone does not enable complete DLQ handling. Also distinguish mapping failures from converter, content format, and HTTP publishing errors; this policy does not cover every error type. Do not use `skip` to hide business events that must be retained. Configure alerting, remediation, and replay procedures for the DLQ. ### Why Do I Receive Duplicate Events After a Restart? Publishing and offset commits are not a single atomic operation. Records published before their offsets are committed may be consumed again, and retries after a lost request response can also produce duplicates. Deduplicate using stable business event IDs and retain an event source that supports reconciliation. Do not interpret a single successful recovery or serial publishing configuration as an unconditional guarantee against loss and duplicates.