Skip to main content

​
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.
Replace <diffusion-host>, <diffusion-username>, and <diffusion-password> 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:
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:
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, 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.