Skip to main content

​
Overview

The JR Source Connector invokes the JR command-line tool installed on the Kafka Connect Worker, generates random JSON objects according to the configuration, and writes the results as Kafka Connect SourceRecords to one Kafka Topic. It is suitable for continuously providing controlled mock data in development, testing, and integration environments, and it can also generate records that conform to a target data structure from a custom template. Each Connector execution can generate one or more objects and periodically generate data according to the configured minimum execution interval. By default, messages contain only a Value. A message Key can also be generated from a template field or a separate key template. Generated results support String, Avro, JSON Schema, and Protobuf Converters.

​
Prerequisites

  • JR must be installed on every Kafka Connect Worker that may run the Task, and the Worker must be able to execute jr. If JR is not added to PATH, configure jr_executable_path.

​
License

MIT License.

​
Quick Start

Prepare a Connect Cluster, a Kafka Topic, and a Worker with JR installed, and confirm that the Worker can access the target Topic. For information about creating and managing the Connector, see AutoMQ’s Manage Connectors. The following configuration uses the built-in net_device template and writes generated data to the jr-generated-data Topic:
Replace topic with the actual target Topic. net_device must be a built-in template available to JR on the Worker. To use a custom template, use embedded_template instead. If JR is not in the Worker’s PATH, also set jr_executable_path.

​
Configuration

​
Template

​
template

Selects the name of a built-in JR Value template.
  • Type: string
  • Default: net_device
  • Importance: High
  • Valid values / notes: When embedded_template is empty, Connector startup runs jr list -n and requires this name to appear in the returned template list. When embedded_template is non-empty, the custom template takes precedence and this check is skipped.

​
embedded_template

Loads a custom Value template from a file or HTTP(S) URL accessible to the Worker.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid values / notes: A non-empty value takes precedence over template. Values other than HTTP(S) URLs are treated as file paths. The file or URL must be accessible to every Worker that may run the Task. Newlines in the loaded content are removed before it is passed to JR.

​
Output

​
topic

Specifies the Kafka Topic to which generated records are written.
  • Type: list
  • Default: None
  • Importance: High
  • Valid values / notes: It must resolve to exactly one non-empty Topic. This Connector does not support routing records to multiple Topics through this configuration.

​
Scheduling and Generation

​
frequency

Sets the minimum interval between two generation checks, in milliseconds.
  • Type: long
  • Default: 5000
  • Importance: High
  • Valid values / notes: A positive value is recommended. The Task generates the next batch only when the current time is strictly later than the previous JR command start time plus this value. This is therefore the minimum interval between the start times of two JR commands; the Task does not wait for another full interval after a command finishes. Configuration parsing does not reject non-positive values, which may cause frequent polling.

​
duration

Limits the total time during which the Task may generate data after startup, in milliseconds.
  • Type: long
  • Default: -1
  • Importance: Medium
  • Valid values / notes: A value greater than 1 establishes a deadline. Timing starts when the Task calls start(), and the first generation that satisfies frequency still runs. Omitting this configuration or setting it to a value no greater than 1 does not establish a deadline; values less than 1 are normalized to -1. Therefore, duration=1 does not mean a 1-millisecond run; it means an unlimited run.

​
objects

Sets the number of objects generated each time JR is invoked.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / notes: A value less than 1 is normalized to 1. Increasing this value increases per-batch memory usage and JSON parsing and serialization work.

​
Message Keys

​
key_field_name

Uses a field in the Value template to generate the message Key.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid values / notes: When non-empty, the Connector requests alternating Key JSON and Value JSON output from JR and uses the specified field for key-value substitution. The implementation does not check whether the field exists at startup. If neither this option nor key_embedded_template is configured, messages do not have a Key.

​
key_value_interval_max

Sets the upper bound for integer key values generated by key_field_name.
  • Type: int
  • Default: 100
  • Importance: Medium
  • Valid values / notes: This option applies only when key_field_name is used and key_embedded_template is not configured. A value less than 1 is normalized to 100. Generated values range from 0 through this upper bound.

​
key_embedded_template

Loads a custom Key template from a file or HTTP(S) URL accessible to the Worker.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid values / notes: When non-empty, it takes precedence over key_field_name and key_value_interval_max. The loading rules are the same as for embedded_template, and JR must output alternating Key JSON and Value JSON objects.

​
Runtime

​
jr_executable_path

Specifies the directory containing the jr executable.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid values / notes: Configure a directory, not a full path that includes the filename. The Connector appends the path separator and jr to this directory. When omitted, the Worker uses PATH. JR must be installed and executable on every Worker that may run the Task.

​
Serialization

​
value.converter

Specifies the Kafka Connect Converter used for generated record Values.
  • Type: string
  • Default: org.apache.kafka.connect.storage.StringConverter
  • Importance: Medium
  • Valid values / notes: Supports StringConverter, AvroConverter, JsonSchemaConverter, and ProtobufConverter. The latter three require the corresponding dependencies on the Worker and the Schema Registry settings required by the Converter. Those Schema Registry settings are not configuration options of this Connector.

​
key.converter

Specifies the Kafka Connect Converter used for generated record Keys.
  • Type: string
  • Default: org.apache.kafka.connect.storage.StringConverter
  • Importance: Medium
  • Valid values / notes: This option has an effect only when a message Key is generated. It supports StringConverter, AvroConverter, JsonSchemaConverter, and ProtobufConverter. When using a Schema Converter, provide matching dependencies and Schema Registry settings on the Worker.

​
Kafka Connect Framework

​
connector.class

Specifies the Connector implementation class.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / notes: Must be set to io.jrnd.kafka.connect.connector.JRSourceConnector.

​
tasks.max

Sets the maximum number of Tasks that Kafka Connect requests for the Connector.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / notes: This Connector always returns one Task configuration and supports only one target Topic. Increasing tasks.max does not create multiple JR generation Tasks or automatically partition Source data.

​
Best Practices

​
Generate a Batch of Test Data Within a Limited Time

Applicable business scenario: You need to generate data for a limited period for an integration test, demonstration, or staged data preparation instead of allowing the Source to run continuously. After a deadline is configured, timing starts when the Task starts. Data is still generated during the first poll that satisfies frequency, and no further data is generated after the deadline is reached. Configuration example:
Replace topic with the Topic used for testing. Finite-duration mode is enabled only when duration is greater than 1. It only stops further data generation; it does not mean that the Connector or Task stops automatically. Key notes: duration timing starts when the Task calls start(), not when the first record is generated. If frequency is greater than the remaining time, the first qualifying poll may still execute once. When a shorter preparation window is required, adjust both frequency and duration. Do not use duration=1 as a short-run configuration.

​
Use a Custom Template to Match the Downstream Data Structure

Applicable business scenario: Built-in templates cannot provide the field structure required by downstream testing or integration. You need to load a custom JR template from a local Worker file or an accessible HTTP(S) address and write the generated JSON to one Topic. Configuration example:
Replace embedded_template with a file path or HTTP(S) URL accessible to every Worker. You do not need to set template when using a custom template. A non-empty embedded_template overrides it. Key notes: The Connector removes newlines from the template content before passing it to JR, so the template must remain valid JR JSON template syntax after that transformation. Files and URLs are read when the Connector starts. When deploying across multiple Workers, ensure that the path, network access, and file content are consistent.

​
Monitoring

​
What to Monitor

Monitor the health and status transitions of the Kafka Connect Worker, Connector, and Task. Track generated-record throughput, record polling latency, Source offset commit progress, errors, and retry counts, as well as Worker JVM heap memory, garbage collection, CPU, and thread signals. If error tolerance or a dead-letter queue is enabled in the deployment, also monitor the corresponding error handling and DLQ activity. Check whether JR subprocess failures cause the Task to enter a failed state.

​
Import the Grafana Dashboard

Download the AutoMQ Connect Cluster Dashboard. In Grafana, import this JSON using a data source that collects Kafka Connect and Worker metrics, and select the Worker, Connector, and Task according to the labels used by the actual cluster.

​
Limitations

  • Only one non-empty target Topic can be configured. The Connector cannot route records to multiple Topics.
  • tasks.max does not split generation work across multiple Tasks. The Connector actually returns one Task configuration.
  • JR must be installed and executable on every Worker that may run the Task. jr_executable_path specifies a directory, not the full path to the executable.
  • embedded_template and key_embedded_template depend on Worker access to file or HTTP(S) resources. A path accessible only to the client that submits the configuration cannot be used as a Worker-local path.
  • Only a duration value greater than 1 limits the run duration. Omitting this configuration or setting it to a value no greater than 1 results in an unlimited run, and finite-duration timing starts when the Task starts.
  • The Source offset position is not a JR record ID, timestamp, or content hash. The implementation updates an internal counter by comparing complete high-precision time strings and increments it again when creating a record, so the value may reset or repeat, commonly as 2. Do not treat it as a globally increasing sequence number.
  • Each generation Task invokes JR again to obtain random data. It does not restore the previous random sequence and does not provide Connector-level deduplication, retry, or dead-letter queue policies.

​
FAQ

​
How do I handle a template-not-found error during Connector startup?

When embedded_template is empty, the Connector calls jr list -n to validate template. On the Worker running the Task, use the same JR installation to confirm that the template name exists and that JR is executable, or configure embedded_template with a file or HTTP(S) address accessible to the Worker. A non-empty embedded_template takes precedence over template.

​
Why does the Connector not stop immediately after duration is configured?

Only a duration greater than 1 establishes a deadline, and timing starts when the Task starts. duration=1 does not establish a deadline and therefore behaves as an unlimited run. After the finite duration is reached, the Connector stops generating further data but does not automatically delete or stop the Connector or Task. Use Connect management operations to manage its lifecycle.

​
Why does the Source offset not continuously increase with the record count?

The Connector’s position is not a business record sequence number. The implementation updates an internal counter by comparing complete UTC ISO time strings. When the time changes, it first resets the counter to 1 and then increments it again while creating the SourceRecord. Consecutive records may therefore repeatedly receive position=2 instead of increasing monotonically. Do not use this field as a business-unique identifier. When stable correlation is required, configure a message Key and let downstream applications use business fields.

​
Why do generated records not have a message Key?

The default configuration generates only a Value. To generate a Key, configure key_field_name or provide a separate Key template through key_embedded_template. The latter overrides key_field_name. Also confirm that the Key Converter matches the generated Key structure.

​
Why are multiple generation Tasks not created after increasing tasks.max?

This Connector’s task configuration always returns one Task, and output is limited to one Topic. tasks.max is Kafka Connect’s requested upper limit and does not change this Connector’s task partitioning logic. To adjust generation volume, evaluate the batch size and execution cadence of the single Task by using frequency together with objects.