Skip to main content

​
Overview

The Diffusion Source Connector subscribes to updates from Diffusion Topics that match a selector, converts each received value into a Kafka Connect SourceRecord, and writes it to Kafka. It sits between Diffusion and Kafka and is suitable for bringing real-time state, device data, or application events into Kafka for stream processing, analytics, and downstream services. The key of each Kafka record is the full Diffusion Topic path, and its value comes from Diffusion JSON data. The target Kafka Topic can use a fixed name or ${topic} to generate a name dynamically from the Diffusion Topic path; with dynamic mapping, / in the path is replaced with _. The Connector uses a single Task to subscribe to and transfer data and does not automatically distribute Diffusion Topics across multiple Tasks.

​
Prerequisites

  • To transfer data, Diffusion must contain JSON Topics that match the selector, and the connection account must have permission to establish a session and select and subscribe to those Topics.
  • The target Kafka Topics must comply with naming rules and allow the Connector to write to them. When using dynamic mapping, check the converted Topic names and potential naming collisions in advance.

​
License

This Connector uses the Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka, a Diffusion service, and the JSON Topics to subscribe to, and verify network connectivity and access permissions. For creation and management operations, see AutoMQ’s Manage Connectors. The following configuration collects updates matching the source/app/ path into one Kafka Topic.
Replace the Diffusion connection URL and credentials, and verify that the selector matches only the JSON Topics you intend to ingest. The example writes all matching updates to diffusion-events, while each record key retains the original Diffusion Topic path. Provide the password through a controlled configuration or secret management mechanism, and do not include it in source code, logs, or shared materials. Whether JsonConverter includes Schema information in the JSON output depends on that Converter’s runtime configuration.

​
Configuration

​
Diffusion Connection and Authentication

​
diffusion.url

Specifies the complete connection URL used to establish a Diffusion session.
  • Category: Diffusion connection
  • Type: string
  • Default: None, required
  • Importance: High
  • Valid values / Notes: Specify a URL accepted by the Diffusion client and accessible to the Worker, such as a WebSocket URL. Configuration parsing does not validate the URL syntax, address reachability, or an empty string. An actual connection failure causes the Task to fail during startup.

​
diffusion.username

Specifies the principal username used to connect to Diffusion.
  • Category: Diffusion authentication
  • Type: string
  • Default: None, required
  • Importance: High
  • Valid values / Notes: The account must have permission to establish a session and select and subscribe to the Topics matched by diffusion.selector. Configuration parsing does not validate whether the account exists or has the required permissions.

​
diffusion.password

Specifies the password for the Diffusion principal.
  • Category: Diffusion authentication
  • Type: password
  • Default: None, required
  • Importance: High
  • Valid values / Notes: Provide the actual password through a controlled secret management mechanism to prevent it from appearing in source code, logs, or shared materials. Diffusion checks the password when establishing the connection.

​
Subscription Scope and Topic Mapping

​
diffusion.selector

Specifies the selector for subscribing to Diffusion Topics.
  • Category: Source data selection
  • Type: string
  • Default: None, required
  • Importance: High
  • Valid values / Notes: Use selector syntax supported by Diffusion and match only the JSON Topics you intend to ingest. The selector is passed unchanged to Diffusion for parsing; configuration validation does not check its syntax, matching results, or permissions. If the subscription cannot complete in time during startup, the Task fails to start.

​
kafka.topic

Specifies the mapping pattern from Diffusion Topic paths to Kafka Topic names.
  • Category: Target Topic
  • Type: string
  • Default: None, required
  • Importance: High
  • Valid values / Notes: A fixed name collects all matching updates into one Kafka Topic. ${topic} is replaced with the complete Diffusion Topic path, after which every / in the result is replaced with _. Other tokens are not replaced. The final name must comply with Kafka Topic rules and have write permission. Different paths may map to the same Topic name.

​
Polling and Batching

​
diffusion.poll.interval

Specifies the wait time in milliseconds before each subsequent check when the in-memory queue temporarily contains no records.
  • Category: Polling
  • Type: int
  • Default: 1000
  • Importance: Low
  • Valid values / Notes: Use a value from 0 to 2147483647. A single empty poll waits at most three times, so the idle wait before an empty result is returned may approach three times this value. 0 only removes the waits between checks and does not change the fixed number of checks. A negative value fails at runtime.

​
diffusion.poll.size

Specifies the maximum number of records returned from the in-memory queue by one poll.
  • Category: Batch
  • Type: int
  • Default: 128
  • Importance: Low
  • Valid values / Notes: Use a value from 1 to 2147483647. This value limits only the number returned by one poll; it does not limit the total queue size and is not the Kafka Producer batch.size. A larger value may increase per-batch conversion and in-memory processing. 0 or a negative value cannot produce usable batching behavior.

​
Runtime and Serialization

​
connector.class

Specifies the Diffusion Source Connector implementation class.
  • Category: Kafka Connect framework configuration
  • Type: string
  • Default: None, required
  • Importance: High
  • Valid values / Notes: Use com.diffusiondata.connect.diffusion.source.DiffusionSourceConnector.

​
tasks.max

Specifies the maximum number of Tasks that the Connector can use.
  • Category: Kafka Connect framework configuration
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / Notes: The value must be at least 1. This Connector always creates only one Task. Increasing the value does not split the selector, increase Source parallelism, or change the single-Task behavior, so keep it at 1.

​
tasks.max.enforce

Specifies whether Kafka Connect enforces that the number of Tasks created by the Connector does not exceed tasks.max.
  • Category: Kafka Connect framework configuration
  • Type: boolean
  • Default: true
  • Importance: Low
  • Deprecated: Yes
  • Replacement: None
  • Valid values / Notes: Valid values are true and false. Kafka Connect has deprecated this configuration, so keep its default value. Disabling the limit does not cause this Connector to create multiple Tasks.

​
key.converter

Specifies the serialization Converter for record keys that contain Diffusion Topic paths.
  • Category: Kafka Connect framework configuration
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: When omitted, this setting inherits the Worker’s Key Converter. The record key is the complete Diffusion Topic path as a string. Use org.apache.kafka.connect.storage.StringConverter when record keys should be serialized as plain text; this class is not the default value.

​
value.converter

Specifies the serialization Converter for record values converted from Diffusion JSON.
  • Category: Kafka Connect framework configuration
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: When omitted, this setting inherits the Worker’s Value Converter. The Connector first converts the JSON value into Kafka Connect data and a dynamically inferred Schema. Use org.apache.kafka.connect.json.JsonConverter when JSON output is required; this class is not the default value. Converter-specific settings are managed by the selected Converter and runtime environment.

​
Best Practices

​
Split Target Kafka Topics by Diffusion Path

Applicable scenario: You have completed initial connectivity validation with a fixed Kafka Topic and now need data from different Diffusion Topics to enter different Kafka Topics. This lets downstream systems configure retention policies, permissions, or consumption jobs separately for each business stream while preserving a clear correspondence between source paths and target names. Configuration example: Starting from the Quick Start, change kafka.topic to the following pattern and leave all other settings unchanged.
Key considerations: For example, the Diffusion path source/app/orders maps to diffusion_source_app_orders. Before deployment, list the paths that the selector may match, verify that the converted names are valid and do not conflict, and prepare the corresponding Kafka Topics or automatic creation policy and write permissions. The mapping replaces only ${topic} and /; it does not remove other invalid characters. foo/bar and foo_bar may produce the same result. If the business needs to collect all updates in one Topic, continue using the fixed name from the Quick Start instead of also trying to split Topics dynamically.

​
Adjust the Wait Time and Batch Size During Operation

Applicable scenario: The Connector is transferring data reliably, and you need to balance the wait time for new data during idle periods, the amount processed per poll, and Worker memory pressure. Configuration example: Override the following two settings in the existing Connector configuration, leaving the connection, selector, Topic mapping, and Converter settings unchanged.
Key considerations: The example values are starting points for tuning, not generally optimal values. A shorter wait time causes more frequent checks of an empty queue. A larger batch size allows more queued records to be returned at once but increases per-batch conversion and in-memory processing. Tune the settings based on Kafka write throughput, end-to-end latency, Worker heap memory, and GC, and continuously monitor backlog trends. These settings do not add a capacity limit to the in-memory queue or provide backpressure to Diffusion.

​
Monitoring

​
What to Monitor

Monitor Kafka Connect cluster health, Connector and Task status, record throughput, end-to-end latency, offset commits, errors and retries, and Worker JVM memory, GC, and thread signals. A Task in the RUNNING state does not mean that the selector is continuously receiving updates; evaluate Diffusion source data changes together with actual Kafka output. Because the Connector’s internal queue has no capacity limit, pay particular attention to heap memory growth under sustained traffic. Monitor DLQ activity only when the deployment has enabled the corresponding error handling.

​
Import the Grafana Dashboard

Download the shared AutoMQ Connect Cluster Dashboard, verify that the Grafana data source can query Kafka Connect and Worker JVM metrics and that cluster, Connector, Task, and other labels match the dashboard requirements, then use Grafana’s import function to load the JSON and select the corresponding data source.

​
Limitations

  • A Connector instance always has only one Task. tasks.max cannot split the selector or increase parallelism.
  • SourceRecord does not contain a recoverable source offset. The Connector cannot restore the Diffusion update position from a Kafka Connect checkpoint or replay updates received while the Task was stopped.
  • The Connector does not provide a basis for claiming exactly-once, at-least-once, or duplicate-free delivery guarantees. Evaluate its use based on replay requirements and downstream idempotency requirements.
  • An unbounded in-memory queue is used between the Diffusion callback and poll, with no source-side backpressure or total backlog limit. Memory usage may continue to grow when sustained input is faster than Kafka delivery.
  • A value Schema is inferred independently for each JSON message. Changes to the data shape may change or remove the Schema, and JSON objects are not restored as predefined Connect Structs.
  • Topic mapping replaces only / with _; it does not remove other invalid characters or detect naming collisions.
  • The Connector does not enable automatic client reconnection when creating a Diffusion session and has no built-in reconnection and resubscription loop.

​
FAQ

​
What should I do if the Task is running normally but no new messages appear in the Kafka Topic?

First, verify that new JSON Topic value updates are actually occurring in Diffusion and that diffusion.selector matches the corresponding paths. Then check the principal’s selection and subscription permissions, the final kafka.topic name, Kafka write permissions, and the Key and Value Converters. A selector with no matches does not cause an error during configuration validation. When using a fixed Topic, do not mistakenly check a dynamically derived name; when using dynamic mapping, look up the path conversion result. Check Worker logs for subscription, conversion, and Kafka write errors, but do not output the actual password.

​
Why does data from different Diffusion Topics enter the same Kafka Topic?

If kafka.topic is a fixed name, data matching the selector is collected in that Topic. To split data by source, use a pattern containing ${topic} and check the conversion results in advance. Even with a dynamic pattern, foo/bar and foo_bar collide because / is replaced with _. Adjust the source path plan or add a non-conflicting fixed prefix, and verify each final Topic name.

​
Why are updates from the downtime not restored after a restart?

The Connector does not save a source offset that can restore the Diffusion update position. On restart, it establishes a new session and resubscribes to the selector, and it cannot rely on Kafka Connect offsets to replay data from the downtime. Before maintenance, evaluate whether the source can retain or replay the required events through another mechanism, and ensure that downstream systems can handle values that may appear more than once. Do not treat a Connector restart as a backfill mechanism.

​
What should I do if the Task does not recover automatically after the Diffusion connection is interrupted?

Check the Diffusion service, network, account permissions, and session URL. After resolving the failure, use the Kafka Connect management interface to confirm the Task status and perform a controlled restart. This Connector has no built-in automatic reconnection and resubscription loop, so do not only wait for the Task to recover by itself. Before restarting, also evaluate the impact of being unable to restore data from the interruption through Connector offsets.