> ## 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. # Shell Sink Connector > Configure and operate the Shell Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## Overview The Shell Sink Connector consumes records from Kafka topics and starts a local shell process for each record on the Connect Worker hosting the Task. In `shell.command`, the Connector replaces `${key}`, `${topic}`, and `${value}` with the current record's key, topic, and value, then executes the complete substituted command through `sh -c`. If the key or value is `null`, the corresponding placeholder is replaced with an empty string. A single Task executes commands one record at a time in the order received within each batch, processing the next record only after the previous command exits. Multiple Tasks can execute concurrently across different partitions. This Connector is suitable for lightweight automation that uses controlled commands and controlled inputs. Placeholder substitution does not perform shell escaping, and the complete substituted command is written to the INFO log, so the Connector is not suitable for directly interpolating untrusted messages or sensitive data. ## Prerequisites * Every Connect Worker host or container that may run a Task must provide `sh` and have the executables, fixed scripts, file paths, and permissions required by the command. Local files and other side effects produced by the command reside on the Worker where the Task actually runs. ## License Uses Apache License 2.0. ## Quick Start Prepare a Connect Cluster, Kafka, and an input topic, and confirm that the Connect Worker hosting the Task can write to `/tmp`. For preparation and management, see [Manage Connectors](../manage-connectors). ```properties theme={null} connector.class=uk.co.threefi.connect.shell.ShellSinkConnector topics= shell.command=printf '%s\\n' processed >> /tmp/shell-sink-events.log ``` Replace `` with the Kafka topic to consume. This fixed command does not interpolate the message key or value. For each consumed record, it appends a line to `/tmp/shell-sink-events.log` on the Worker hosting the current Task; each line contains `processed`, which can be used to confirm the basic execution path. The complete substituted command is written to the Worker's INFO log. For production use, replace it with a pre-deployed, permission-restricted fixed program or script that has an explicit execution timeout; do not directly interpolate untrusted or sensitive message content. ## Configuration ### Command Execution #### `shell.command` The shell command template executed for each Kafka record. * **Type**: `string` * **Default**: Empty string * **Importance**: High * **Valid Values / Notes**: Supports `${key}`, `${topic}`, and `${value}`. The Connector first performs text substitution and then runs the complete command through `sh -c`. Placeholder values are neither quoted nor escaped, and the complete substituted result is written to the INFO log. An empty string passes configuration parsing but produces no useful business side effect. The command runs in the environment, current directory, file system, and permission context of the Worker hosting the Task. Each record starts a separate process, and the Connector does not provide an execution timeout. ### Failure Retries #### `max.retries` The number of Connector-managed retries allowed after a shell command fails. * **Type**: `int` * **Default**: `0` * **Importance**: Medium * **Valid Values / Notes**: The minimum value is `0`. `0` causes the Task to fail immediately after the first command failure. A positive integer `N` allows at most `N` retries in addition to the initial attempt. A retry re-executes the entire Worker batch, not only the failed record. After a batch succeeds, the Task's retry budget resets to the configured value. #### `retry.backoff.ms` The upper bound, in milliseconds, passed to Kafka Connect for the next `consumer.poll` after a retryable command failure. * **Type**: `int` * **Default**: `3000` * **Importance**: Medium * **Valid Values / Notes**: The minimum value is `0`. This setting is used only when `max.retries` is greater than `0` and retry budget remains. In Kafka Connect 3.9.1, the value is the maximum time the next `consumer.poll` may block. It may be shortened further by the remaining time until the next offset commit, and the poll may also return early, so it does not guarantee waiting at least this long before the actual retry. This setting is not a shell command execution timeout. ### Connector and Tasks #### `connector.class` The Shell Sink Connector implementation class to load. * **Type**: `string` * **Default**: None * **Importance**: High * **Valid Values / Notes**: Use `uk.co.threefi.connect.shell.ShellSinkConnector`. * **Required**: Yes #### `tasks.max` The maximum number of Sink Tasks allowed to start. * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid Values / Notes**: The minimum value is `1`. Actual parallelism is also limited by the number of input topic partitions and Worker assignment. Different Tasks can execute commands concurrently, so the target script and external system must support concurrent calls. The Connector does not provide global execution ordering across partitions or Tasks. ### Kafka Input #### `topics` The list of Kafka topics to consume. * **Type**: `list` * **Default**: `[]` * **Importance**: High * **Valid Values / Notes**: Separate multiple topics with commas. Configure exactly one of `topics` and `topics.regex`. * **Required**: Conditionally required #### `topics.regex` A Java regular expression that dynamically matches input topics. * **Type**: `string` * **Default**: Empty string * **Importance**: High * **Valid Values / Notes**: Must be a valid Java regular expression. Configure exactly one of `topics.regex` and `topics`. * **Required**: Conditionally required ### Record Conversion #### `key.converter` The Converter class that deserializes the Kafka message key before `${key}` is substituted. * **Type**: `class` * **Default**: Not set (`null`) * **Importance**: Low * **Valid Values / Notes**: When not set, this setting inherits the Worker-level key Converter. The Connector calls Java `toString()` on the converted key, and a `null` key is replaced with an empty string. The Connector does not automatically shell-escape Converter output. #### `value.converter` The Converter class that deserializes the Kafka message value before `${value}` is substituted. * **Type**: `class` * **Default**: Not set (`null`) * **Importance**: Low * **Valid Values / Notes**: When not set, this setting inherits the Worker-level value Converter. The Connector calls Java `toString()` on the converted value, and the `null` value of a tombstone record is replaced with an empty string. Structured values such as maps, structs, arrays, and byte arrays use the string form of their corresponding Java objects. They are not guaranteed to preserve the original JSON, CSV, or binary representation and are not automatically shell-escaped. ## Best Practices ### Use Bounded Retries for Transient Failures **Applicable scenario:** Initial integration is complete, but a fixed script occasionally returns a nonzero exit code because of temporary resource contention or target service unavailability. The external operation performed by the command must be safe to repeat, and the Task should retry automatically a limited number of times instead of stopping after the first failure. **Configuration example:** ```properties theme={null} connector.class=uk.co.threefi.connect.shell.ShellSinkConnector topics= shell.command=/opt/connect/bin/process-shell-event max.retries=3 retry.backoff.ms=5000 ``` **Key notes:** Deploy `/opt/connect/bin/process-shell-event` as a permission-restricted fixed script on every candidate Worker, and have the script enforce an explicit execution timeout. `max.retries=3` allows at most three retries in addition to the initial attempt; the Task fails when the budget is exhausted. `retry.backoff.ms=5000` requests a maximum blocking timeout of 5000 milliseconds for the next poll, but does not guarantee waiting at least 5000 milliseconds before the actual retry. Retries operate on the entire Worker batch: if records at the start of a batch have already completed successfully and a later record fails, the successful prefix is executed again. Enable retries only when the script and target system provide idempotency or deduplication. This configuration does not provide exactly-once guarantees. ### Scale Tasks by Partition When Backlog Grows **Applicable scenario:** The Connector is running reliably, the input topic has multiple partitions, serial command execution by a single Task causes a sustained backlog, and the fixed script and target system can handle concurrent calls. Gradually increase the Task limit to process records in parallel across partitions. **Configuration example:** ```properties theme={null} connector.class=uk.co.threefi.connect.shell.ShellSinkConnector topics= tasks.max=4 shell.command=/opt/connect/bin/process-shell-event ``` **Key notes:** `tasks.max=4` is only an upper limit on the number of Tasks; effective parallelism cannot exceed the number of input partitions. Each Task still executes records one at a time in batch order, but different Tasks can run concurrently on the same or different Workers, with no global ordering guarantee across partitions. Before scaling, confirm that the script is consistently available on every candidate Worker and that target files, locks, and external services support concurrency. Adjust the value gradually based on backlog, command duration, and target capacity instead of treating the example value as a fixed optimum. ## Monitoring ### What to Monitor Monitor Kafka Connect health, Connector and Task status, throughput, latency, offset commits, errors, retries, and Worker JVM signals. Also monitor Tasks that make no progress for extended periods and external command duration to prevent commands without timeouts from blocking consumption indefinitely. Monitor DLQ activity only when the corresponding error handling is enabled. ### Import the Grafana Dashboard Ensure that Connect metrics are available in a Grafana data source and that the collected labels match the dashboard filters; download the [Kafka Connect Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), import the JSON into Grafana, and select the corresponding data source. ## Limitations * `${key}`, `${topic}`, and `${value}` are unescaped text substitutions that are subsequently parsed by `sh -c`. Directly placing untrusted record content in a command can change command syntax and cause command injection. * The complete substituted command is written to the Worker's INFO log. Command templates and placeholder content must not contain passwords, tokens, credentials, or other sensitive data. * The Connector does not provide a shell command execution timeout and cannot actively terminate a running child process when a Task stops. A command that does not exit blocks the Task indefinitely. * The Connector does not support exactly-once semantics and does not include shell side effects and Kafka offset commits in the same transaction. Whole-batch retries and failures after a command succeeds but before the offset is committed can both cause duplicate execution. * Each record starts a separate process. Commands execute serially within a single Task, so long-running commands and process startup overhead directly limit throughput. * `${key}` and `${value}` use the Java string form of the object produced by the Converter. The Connector does not provide JSON, CSV, or binary encoding, or field-path templates. * Kafka Connect's `errors.tolerance` and DLQ mechanisms cannot skip or receive shell command failures that occur inside `SinkTask.put`. The Connector only uses `max.retries` and `retry.backoff.ms` to perform bounded retries of the entire batch, and the Task fails when the retry budget is exhausted. ## FAQ ### Why Does the Task Stop After a Command Returns a Nonzero Exit Code? The default value of `max.retries` is `0`, so the first command failure immediately produces an unrecoverable Task error. Check the Worker log for the Task, the command exit code, script path, execution permissions, and required programs, then reproduce the command using the same user and environment as the Worker. Configure bounded `max.retries` and `retry.backoff.ms` only when the failure is genuinely transient and the external operation can be repeated safely. After correcting the problem, restart the failed Task through the Connect management process. ### Why Does the Target System Show Duplicate Operations? A retry after a command failure resubmits the entire Worker batch, so records that already completed successfully within the batch may execute again. Even when the entire batch succeeds, records can be replayed if the Worker fails before committing the Kafka offset. Do not treat a Kafka offset commit as transactional confirmation of shell side effects. Prefer idempotent commands and target-side deduplication, and ensure that deduplication identifiers come from controlled input that has been strictly validated, cannot alter shell syntax, and does not expose sensitive information. ### Why Is the Task Still Running but Not Making Consumption Progress? A command may run for a long time, wait for input, or block on output, while the Connector provides no command timeout or active termination mechanism. First inspect the corresponding child process and script logs on the Worker hosting the Task. Then correct the fixed program or script so that it does not depend on interactive input, limits its output, and enforces explicit connection and execution timeouts. After handling any remaining processes, restart the Task and continue monitoring consumption latency and offset progress. ### Can Quoting Make Message Values Safe? Adding single or double quotes around a placeholder in the template is not a general safeguard. Placeholders are substituted verbatim before execution, so quotes, command substitutions, redirection operators, newlines, or other shell metacharacters in a message can still break out of the intended syntax. Do not directly interpolate untrusted keys or values. If record content must be used, restrict input to a strictly defined set of allowed characters before it reaches the Connector, avoid sensitive data, and use a fixed command boundary that has undergone a security review.