> ## 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. # Diffusion Sink Connector > Configure and operate the Diffusion Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## Overview The Diffusion Sink Connector writes records from Kafka topics to a Diffusion Server. It converts each record value to JSON that Diffusion can process and resolves the target Diffusion topic path from the configuration. If the target topic does not exist, the Connector creates it as a JSON topic before writing the value; otherwise, it updates the existing topic. The Connector can map one or more Kafka topics to a fixed Diffusion path, or use `${topic}`, `${key}`, `${key.version}`, and `${value.version}` to generate paths from record metadata. A typical use case is publishing business events from Kafka to Diffusion for real-time consumption by browser, mobile, and IoT clients. ## Prerequisites * Diffusion Server 6.9 or later, with a service account prepared to receive data; the account must be able to update the target path and create JSON topics when the target topic does not exist. * A runtime using Java 11 or later, with access to the Diffusion Server through `diffusion.url`. ## License Uses Apache License 2.0. ## Quick Start Prepare a Connect Cluster, Kafka, and Diffusion Server in advance, and confirm network connectivity and service account permissions. For specific preparation and management operations, see [Manage Connectors](../manage-connectors). ```properties theme={null} connector.class=com.diffusiondata.connect.diffusion.sink.DiffusionSinkConnector topics=price diffusion.url=ws://:8080 diffusion.username= diffusion.password= diffusion.destination=kafka/${topic} ``` Replace ``, ``, and `` with the actual connection details. This configuration writes records from the Kafka topic `price` to the Diffusion topic `kafka/price`. Submit only the Connector properties shown above, without a REST request wrapper. ## Configuration ### Diffusion Connection #### `diffusion.url` The connection address of the Diffusion Server. * **Type**: `string` * **Default**: None * **Importance**: High * **Valid values / notes**: Required. Use a complete address that the Kafka Connect Worker can access; the Connector configuration parses this value as a string only and does not validate the protocol, host, port, or whether the value is empty. * **Required**: Yes #### `diffusion.username` The principal used to authenticate with the Diffusion Server. * **Type**: `string` * **Default**: None * **Importance**: High * **Valid values / notes**: Required. The account must have permission to access the target Diffusion Topic; the Connector configuration does not validate whether the value is empty. * **Required**: Yes #### `diffusion.password` The password used to authenticate with the Diffusion Server. * **Type**: `password` * **Default**: None * **Importance**: High * **Valid values / notes**: Required. Store it using Connect's sensitive configuration management; do not write the password to logs or public configuration repositories. * **Required**: Yes ### Target Path #### `diffusion.destination` The pattern used to generate the Diffusion Topic path. The Connector resolves this pattern for each SinkRecord. * **Type**: `string` * **Default**: None * **Importance**: High * **Valid values / notes**: Required. Supports `${topic}`, `${key}`, `${key.version}`, and `${value.version}`. `${topic}` uses the input Kafka Topic; the other tokens depend on the record key or Schema version. When the record key is null, the Connector does not continue resolving key and version tokens after replacing `${topic}`; tokens without available values remain as literals. Path separators are not cleaned or replaced. * **Required**: Yes ### Input Subscription #### `topics` The list of Kafka topics to consume. * **Type**: `list` * **Default**: empty list * **Importance**: High * **Valid values / notes**: Mutually exclusive with `topics.regex`. Use comma-separated topic names and configure at least one non-empty topic. Topic names can participate in target path generation through `${topic}`. #### `topics.regex` Subscribes to Kafka topics using a Java regular expression. * **Type**: `string` * **Default**: empty string * **Importance**: High * **Valid values / notes**: Mutually exclusive with `topics`. The value must be a compilable, non-empty Java regular expression. If a DLQ topic is configured, the expression must not match that topic. ### Connector Identity and Tasks #### `connector.class` The Sink Connector implementation class to load. * **Type**: `string` * **Default**: None * **Importance**: High * **Valid values / notes**: Use `com.diffusiondata.connect.diffusion.sink.DiffusionSinkConnector`. * **Required**: Yes #### `tasks.max` The maximum number of tasks that this Connector may use. * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid values / notes**: Must be greater than or equal to `1`. This implementation always returns one task configuration, so increasing this value does not cause a single Connector instance to create multiple Diffusion Sink tasks. #### `tasks.max.enforce` Whether to enable Kafka Connect's framework constraint for `tasks.max`. * **Type**: `boolean` * **Default**: `true` * **Importance**: Low * **Valid values / notes**: Optional. This configuration is deprecated in Kafka Connect 3.9.1 and has no replacement; it does not change the Connector's one-task behavior. * **Deprecated**: Yes ### Data Conversion #### `key.converter` Specifies the Converter for Kafka record keys. The converted key can be used with `${key}`, and its Schema version can be used with `${key.version}`. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: Optional. When omitted, inherits the Worker's key Converter; when configured, it must be a concrete Converter class that can be instantiated. #### `value.converter` Specifies the Converter for Kafka record values. The converted value is serialized as Diffusion JSON, and its Schema version can be used with `${value.version}`. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: Optional. When omitted, inherits the Worker's value Converter; when configured, it must be a concrete Converter class that can be instantiated. Map keys must be representable as strings; values that do not meet this requirement cause record processing to fail. ### Error Handling #### `errors.retry.timeout` The total retry duration for retryable errors, in milliseconds. * **Type**: `long` * **Default**: `0` * **Importance**: Medium * **Valid values / notes**: `0` disables retries, `-1` enables infinite retries, and other non-negative values specify the retry duration. This is a general Kafka Connect policy and does not replace the fixed wait behavior used by the Sink Task while waiting for Diffusion publication results. #### `errors.tolerance` The scope of continued processing allowed when an error occurs. * **Type**: `string` * **Default**: `none` * **Importance**: Medium * **Valid values / notes**: Valid values are `none` and `all`. This is a general error policy; configuring it alone does not guarantee that Diffusion publication failures are automatically written to the DLQ. #### `errors.deadletterqueue.topic.name` The Kafka DLQ topic name used to store error records. * **Type**: `string` * **Default**: empty string * **Importance**: Medium * **Valid values / notes**: An empty string means that records are not published to a DLQ. When a non-empty value is configured, that topic must not also appear in `topics` or match `topics.regex`. #### `errors.deadletterqueue.topic.replication.factor` The replication factor used when Kafka Connect creates the DLQ topic. * **Type**: `short` * **Default**: `3` * **Importance**: Medium * **Valid values / notes**: Used only when the configured DLQ topic does not exist and Kafka Connect creates it; the value must be suitable for the number of replicas in the Kafka cluster. #### `errors.deadletterqueue.context.headers.enable` Whether to add error context headers to records written to the DLQ. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid values / notes**: Takes effect only when DLQ records are actually produced; when enabled, the error context uses the `__connect.errors.` prefix. ## Best Practices ### Connect Kafka Data Streams by Topic Pattern **Applicable business scenario**: You need to continuously ingest a set of Kafka topics whose names follow a consistent pattern and write each topic's records to the corresponding Diffusion path. A regular-expression subscription avoids maintaining the topic list individually. **Configuration example**: ```properties theme={null} connector.class=com.diffusiondata.connect.diffusion.sink.DiffusionSinkConnector topics.regex=orders-.* diffusion.url=ws://:8080 diffusion.username= diffusion.password= diffusion.destination=events/${topic} ``` **Key notes**: Do not configure a non-empty `topics` when `topics.regex` is non-empty. `${topic}` preserves the input topic name, so `orders-created` is written to `events/orders-created`. Before expanding the subscription scope, confirm that the Diffusion principal has update or create permission for every target path that may be generated. ### Separate Data Streams with Multiple Connector Instances **Applicable business scenario**: You need to manage business streams separately, assign different target paths or failure boundaries, and the topic set in one Connector instance has become large. Create an independent Connector instance for each topic group. **Configuration example**: ```properties theme={null} connector.class=com.diffusiondata.connect.diffusion.sink.DiffusionSinkConnector topics=orders,inventory tasks.max=1 diffusion.url=ws://:8080 diffusion.username= diffusion.password= diffusion.destination=business/${topic} ``` When creating a second Connector instance for another topic group, use different `topics` or `topics.regex` values and a different target path instead of only increasing `tasks.max` on the same instance. This implementation always creates one task, so increasing `tasks.max` does not create task-level parallelism within an instance. Divide input topics explicitly between instances to avoid consuming the same data stream more than once. ## Monitoring ### What to Monitor Monitor Kafka Connect health, Connector and task status, throughput, latency, offset commits, errors, retries, and Worker JVM signals. Monitor DLQ activity only when the corresponding error handling is enabled. ### Import the Grafana Dashboard Confirm that Connect metrics are connected to a Grafana data source and that the collected labels meet the dashboard's filtering requirements; download the [Kafka Connect Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), import the JSON in Grafana, and select the corresponding data source. ## Limitations * A single Diffusion Sink Connector instance always creates one task; increasing `tasks.max` does not make the instance create multiple tasks in parallel. To scale at the deployment level, separate input topics across multiple instances. * Exactly one non-empty subscription method must be selected between `topics` and `topics.regex`; if both are non-empty or both are empty, Sink configuration validation fails. * If the connection is lost or the task restarts after a Diffusion update succeeds but before the framework commits the Kafka offset, the offset may remain uncommitted. The record may be written to the same path again after recovery, so the Connector does not guarantee that application-level side effects occur only once. * When the record key is null, `diffusion.destination` replaces only `${topic}`; `${key}`, `${key.version}`, and `${value.version}` are not resolved further. Therefore, a path pattern that depends on these tokens should ensure that input records have the required key and Schema version. ## FAQ ### Why is there still only one task after increasing `tasks.max`? This is the Connector's implementation behavior. Its `taskConfigs` method always returns one task configuration, and `tasks.max` only sets Kafka Connect's task limit; it does not change the number of tasks in a single instance. To isolate data streams or scale the deployment, create multiple Connector instances for different topic sets and ensure that they do not subscribe to the same topic. ### Why can't the Connector start after configuring both `topics` and `topics.regex`? A Kafka Connect sink requires exactly one of them to be non-empty. Remove one, or change `topics.regex` to a valid, non-empty Java regular expression. If a DLQ topic is also configured, confirm that it is not included in the subscription list or matched by the regular expression. ### Why did unexpected topic paths appear in Diffusion? Check whether the tokens in `diffusion.destination` match the key and Schema version actually available in the records. `${topic}` uses the input Kafka topic, and `${key}` uses a non-null record key. When the record key is null, version tokens are not resolved; a token can also remain literal when its value is unavailable. Start with a fixed path or `${topic}` alone to verify the input scope, then add key and version tokens incrementally. ### Why is the same record processed again after a successful publication and restart? There is a window between a successful Diffusion update and the Kafka offset commit. If the connection is interrupted after the update completes but before the offset is committed, Kafka Connect may redeliver the record during recovery. Check whether the target uses the same key and path, assess whether the application can tolerate duplicate processing, and do not treat Diffusion's last-write result as an application-level idempotency guarantee. ### Why did the error not enter the DLQ? The DLQ is a general Kafka Connect error-handling capability, and DLQ records are produced only when `errors.deadletterqueue.topic.name` is configured and the applicable error-handling conditions are met. Diffusion publication failures are handled by the Sink Task's publication Future and flush wait; configuring only `errors.tolerance` does not guarantee that these failures are automatically written to the DLQ. First inspect Connector, Task, error, and retry metrics and logs to determine whether the failure is a record-level conversion error or a target-side publication error.