> ## 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. # JR Source Connector > Configure and operate the JR Source Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## 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](../manage-connectors). The following configuration uses the built-in `net_device` template and writes generated data to the `jr-generated-data` Topic: ```properties theme={null} connector.class=io.jrnd.kafka.connect.connector.JRSourceConnector template=net_device topic=jr-generated-data ``` 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**: ```properties theme={null} connector.class=io.jrnd.kafka.connect.connector.JRSourceConnector template=net_device topic=jr-test-data frequency=5000 duration=60000 objects=10 ``` 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**: ```properties theme={null} connector.class=io.jrnd.kafka.connect.connector.JRSourceConnector embedded_template=/opt/jr/templates/device.json topic=jr-custom-data frequency=10000 objects=5 ``` 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](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json). 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`.