Overview
Shell Source Connector runs shell commands on the Kafka Connect Worker hosting the Task, converts each line of standard output into a Kafka record, and writes it to the specified Topic. It connects external data sources to Kafka, ingesting output from existing command-line tools, commands that read finite text files, or collection scripts. After each command completes successfully, the Connector returns its output to Kafka Connect as a batch, then runs the command again on the next poll. Each record value is a string with the line terminator removed, the record key is null, and the timestamp is the time the record is constructed. The command determines which data to collect and whether to output only new data; the Connector does not automatically detect new file content or parse JSON or CSV fields.Prerequisites
- An executable
shmust be available in the operating system or container of every candidate Worker that may run the Task. - Tools, scripts, and source resources used by the command must be available on every candidate Worker and accessible to the Worker process; paths in containers must refer to resources inside the container.
- The command must finish within a finite amount of time, and its text output per execution should be bounded, because the entire output is handed to Kafka Connect only after the command finishes.
License
Uses Apache License 2.0.Quick Start
Prepare a Connect Cluster, Kafka, and the target Topic in advance, confirm network connectivity and access permissions, and ensure that the Worker can executesh. For preparation and management operations, see AutoMQ’s Manage Connectors. The following command uses the Shell’s printf to output two fixed lines of text; no external file is required.
<target-topic> with the target Topic name and apply the settings above to the Connector configuration. Each successful collection produces two text records with values alpha and beta, using one Task by default. The command runs repeatedly, so these text values continue to appear rather than being sent only once. The double backslash in the code is for reading a properties file; after parsing, the command’s format string is %s\n.
Configuration
Connector and Execution Concurrency
connector.class
Selects the Shell Source Connector implementation.
- Category: Kafka Connect framework configuration
- Type:
string - Default: None
- Importance: High
- Required: Yes
- Valid Values / Notes: Use
uk.co.threefi.connect.shell.ShellSourceConnector.
tasks.max
The maximum number of Tasks allowed for the Connector.
- Category: Kafka Connect framework configuration
- Type:
int - Default:
1 - Importance: High
- Valid Values / Notes: Values range from
1to2147483647. Every Task receives the same command configuration; the source data scope is not automatically divided. For a single source and a single command, generally keep the default. Increasing the number of Tasks may cause duplicate execution or resource contention rather than partitioning the data.
tasks.max.enforce
Whether to enforce the tasks.max limit.
- Category: Kafka Connect framework configuration
- Type:
boolean - Default:
true - Importance: Low
- Deprecated: Yes
- Replacement: None
- Valid Values / Notes: Accepts
trueorfalse. Keep the default and control the number of Tasks throughtasks.max; do not rely on disabling this check to increase parallelism.
Command and Collection Timing
shell.command
The shell command string executed on each poll.
- Category: Connector configuration
- Type:
string - Default: Empty string
- Importance: High
- Valid Values / Notes: Explicitly specify a valid command that can finish successfully. The command is executed by
sh -con the Worker, inheriting the Worker process’s environment and working directory; use absolute paths for files or scripts where possible. Only stdout is collected, and it is decoded using the Worker JVM’s default character set. Command text may appear in logs, so do not include passwords, tokens, or other credentials. Template substitution based on message keys, Topics, or values is not supported.
block.ms
The wait time, in milliseconds, after each command has been processed and before the collection batch is returned.
- Category: Connector configuration
- Type:
int - Default:
100 - Importance: Low
- Valid Values / Notes: Use an integer from
0to2147483647. Configuration parsing does not validate that the value is nonnegative, but the actual wait requires a nonnegative value. The wait occurs after successful execution, empty output, or a caught command-execution I/O error; output from the first collection is also handed to Connect only after the wait. This is not a command timeout, a fixed interval, or cron scheduling; the interval between collections also includes command execution and Worker processing time.
Output Destination and Text Serialization
topic
The target Kafka Topic for all output lines.
- Category: Connector configuration
- Type:
string - Default: Empty string
- Importance: High
- Valid Values / Notes: Explicitly specify one valid Topic name, not a list of names or a regular expression. Plugin configuration validation does not check whether the Topic is valid or whether access permissions are available.
value.converter
Specifies the Converter used when writing Source record values to Kafka.
- Category: Kafka Connect framework configuration
- Type:
class - Default:
null - Importance: Low
- Valid Values / Notes: When omitted, inherits the Worker’s value Converter. Use
org.apache.kafka.connect.storage.StringConverterwhen writing plain text to prevent the output format from varying with the Worker’s default settings. The Converter must be an instantiable Kafka Connect Converter implementation; selecting a Converter does not make the plugin automatically parse JSON or CSV strings into fields.
Best Practices
Adjust the Wait Time to Match Source Load During Continuous Collection
Applicable Scenario: After initial setup, you need to keep running a collection command that finishes on each execution, but frequent queries or script runs increase the load on the source. A longer wait after each collection reduces access frequency at the cost of higher delivery latency. Configuration Example: Add the following setting to the running Connector’s configuration, or override the existingblock.ms; keep the original command, Topic, and StringConverter.
tail -f; increasing the wait does not resolve repeated output or provide a dedicated backoff strategy after failures.
Monitoring
What to Monitor
Monitor Kafka Connect cluster health, Connector and Task status, record throughput and end-to-end latency, Offset commit success rates and duration, errors and retries, and Worker JVM memory, GC, and thread signals. A healthy Task status does not mean every command execution succeeds; assess source execution results together with the actual Kafka output. Successful Offset commits also do not mean collection can resume from a saved source position. Monitor DLQ activity only when the corresponding error handling is enabled in the deployment; do not assume command execution failures are record errors that automatically go to the DLQ.Import the Grafana Dashboard
Download the shared Kafka Connect Grafana Dashboard, confirm that the monitoring data source collects Kafka Connect metrics and that labels such as cluster, Worker, Connector, and Task match the dashboard queries, then use Grafana’s import feature to upload the JSON and select the corresponding data source.Limitations
- Collects only stdout text; it does not automatically collect stderr or provide binary passthrough.
- Output is returned only after the command finishes with a zero exit code; commands that run indefinitely cannot be used as real-time, line-by-line streams.
- The entire output of each execution is held in memory, with no setting to split batches by record count or byte size.
- File positions, line numbers, and source cursors are neither saved nor restored; committed timestamp Offsets cannot be used to resume reading from a saved position.
- Does not automatically deduplicate records or atomically tie source archiving, deletion, or external cursor advancement to Kafka writes.
- Does not provide message-data template substitution, built-in business-key extraction, or dynamic Topic routing.
- Does not provide a command execution timeout setting or a plugin mechanism to actively terminate child processes.
- Does not guarantee global record ordering across Kafka partitions or multiple Tasks.
FAQ
What Should I Check When the Task Is Running Normally but the Topic Has No New Messages?
Confirm that the command produces stdout when run in the actual Worker container or host with the Worker’s permissions and environment, and that it finishes with a zero exit code. Check tool paths, script permissions, and source resource locations. Commands that write only to stderr do not produce records; if a command produces output but exits with a nonzero status, its entire output is discarded. Check command execution warnings in the Worker and investigate source-side errors without exposing credentials; after fixing paths, permissions, or the command, observe the output again. Also check the target Topic, Converter, andblock.ms, since a long wait delays the current batch. After a command I/O error is caught, the command runs again in a subsequent collection, so Task status alone is not enough to determine whether source execution succeeded.
Why Does the Same Content Keep Appearing, and Why Does Reading Not Resume from the Next Line After a Restart?
The Connector executes the entire command each time and does not read historical Offsets as a source recovery position. Reading the same file or repeatedly querying the same content produces records again, including after a restart. Before resuming after maintenance, confirm that the source data is still replayable, and have downstream systems handle duplicates using business identifiers. If the business requires reliable incremental cursors and archiving after acknowledgment, use a data source or Connector with the corresponding recovery protocol. Do not replace Kafka write acknowledgment by moving or deleting a file immediately after reading it; doing so may remove replayable data before it has been delivered.Why Does Increasing the Number of Tasks Cause Duplicate Collection?
Every Task executes the same command, without dividing work by file, data scope, or source partition. Checktasks.max and restore it to 1 for a single source and a single command; also check whether candidate Workers refer to the same resource. Do not treat additional Tasks as source partitioning. Before scaling, define nonoverlapping data scopes and resource ownership, then design independent collection tasks.