> ## 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.

# Redis Sink Connector

> Configure and operate the Redis Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting.

## Overview

Redis Sink Connector consumes records from a Kafka topic and writes them to Redis according to the configuration. It can write records to strings, hashes, JSON documents, Streams, Lists, Sets, Sorted Sets, or RedisTimeSeries. `redis.command` determines the target data structure and how record keys and values are mapped, while `redis.key` determines the Redis keyspace; collection types use the topic-expanded keyspace, while non-collection types such as strings, hashes, and JSON usually also append the Kafka record key.

This Sink is suitable for synchronizing events, state, or metrics from Kafka to Redis for low-latency queries, caching, collection operations, stream consumption, or time-series use cases. Its delivery semantic is at least once; when choosing Redis commands that append data, design the target keys and business deduplication strategy with possible replays in mind.

## Prerequisites

* When using `JSONSET`, the target Redis deployment must provide RedisJSON; when using `TSADD`, it must provide RedisTimeSeries.
* When using TLS, ACL, or mutual TLS, the Connect Worker must be able to read the configured credentials, CA certificate, client certificate, and private-key files.

## License

Uses Apache License 2.0.

## Quick Start

Prepare the Connect Cluster, Kafka, and Redis in advance, and confirm that the Worker can access the target Redis, consume the specified topic, and use the target Redis commands. For preparation and management operations, see [Manage Connectors](../manage-connectors).

```properties theme={null}
connector.class=com.redis.kafka.connect.RedisSinkConnector
topics=orders
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
redis.host=<redis-host>
redis.port=6379
redis.command=SET
redis.key=${topic}
```

Replace `orders` with the Kafka topic to consume and `<redis-host>` with the Redis address. This example writes string record values to a Redis String; a non-empty record key generates a target key in the form `<topic>:<record-key>`. After submitting the configuration to the Connect Cluster, check the Connector and Task status and confirm that the expected key and value appear in the target Redis.

## Configuration

### Redis Connection

#### `redis.uri`

URI used to create the Redis connection, supporting the `redis://` and `rediss://` schemes.

* **Type**: `string`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / notes**: Takes precedence over `redis.host` and `redis.port` when non-empty; when empty, creates the connection using the host and port.

#### `redis.host`

Redis host address.

* **Type**: `string`
* **Default**: `localhost`
* **Importance**: High
* **Valid values / notes**: Used only when `redis.uri` is empty; specify a hostname or address accessible to the Worker.

#### `redis.port`

Redis TCP port.

* **Type**: `int`
* **Default**: `6379`
* **Importance**: High
* **Valid values / notes**: Used together with `redis.host` only when `redis.uri` is empty; specify a valid Redis TCP port.

#### `redis.cluster`

Whether to use the Redis Cluster client.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / notes**: `true` uses the Cluster client and `false` uses the regular Redis client; do not choose the Cluster client when using `redis.multiexec=true`.

#### `redis.timeout`

Redis command timeout, in seconds.

* **Type**: `long`
* **Default**: `60`
* **Importance**: Medium
* **Valid values / notes**: Used for the Redis client and batch-write waiting; adjust it according to Redis response time and business latency requirements.

#### `redis.pool`

Maximum connection-pool size used by the Redis Writer.

* **Type**: `int`
* **Default**: `8`
* **Importance**: Medium
* **Valid values / notes**: The pool size affects concurrent write capacity; adjust it based on the number of Tasks, Redis connection limits, and Worker resources.

### Authentication and TLS

#### `redis.username`

Username used for Redis ACL authentication.

* **Type**: `string`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / notes**: Use with a non-empty `redis.password`.

#### `redis.password`

Redis password.

* **Type**: `password`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / notes**: Applied to the Redis URI when non-empty; do not write a real password to documentation or logs.

#### `redis.tls`

Whether to enable Redis TLS.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / notes**: `true` enables TLS; configure a trusted CA or client certificate according to the server certificate.

#### `redis.insecure`

Whether to disable peer certificate verification for TLS connections.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / notes**: Effective only when `redis.tls=true`; `true` disables peer verification and is suitable only for environments whose risks have been assessed.

#### `redis.cacert`

X.509 CA certificate file used to verify the Redis server.

* **Type**: `string`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / notes**: Specify a CA certificate file path readable by the Worker; when empty, the default trust configuration is used.

#### `redis.key.file`

PKCS#8 private-key PEM file used for mutual TLS.

* **Type**: `string`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / notes**: Specify a private-key file path readable by the Worker; after setting it, also set the corresponding `redis.key.cert`.

#### `redis.key.cert`

X.509 client certificate-chain PEM file paired with `redis.key.file`.

* **Type**: `string`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / notes**: Provide the corresponding certificate-chain file when setting `redis.key.file`.

#### `redis.key.password`

Password for the `redis.key.file` private key.

* **Type**: `password`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / notes**: Specify the private-key password when the key is encrypted; use an empty value for an unencrypted private key.

### Target Data Mapping

#### `redis.command`

Selects the Redis data structure or operation to write.

* **Type**: `string`
* **Default**: `XADD`
* **Importance**: High
* **Valid values / notes**: Can be `HSET`, `JSONSET`, `TSADD`, `SET`, `XADD`, `LPUSH`, `RPUSH`, `SADD`, `ZADD`, or `DEL`; case must match. `HSET` and `XADD` require Struct or Map values; `ZADD` uses a numeric score as its value; `TSADD` uses the record key as a timestamp in milliseconds and the record value as a numeric sample.

#### `redis.key`

Format string for the target Redis keyspace.

* **Type**: `string`
* **Default**: `${topic}`
* **Importance**: Medium
* **Valid values / notes**: `${topic}` is replaced with the source Kafka topic. Collection types use only the expanded keyspace; other types usually also append the record key using `redis.separator`. When empty, non-collection types can use the Kafka record key directly as the target key.

#### `redis.separator`

Separator between the keyspace and Kafka record key for non-collection targets.

* **Type**: `string`
* **Default**: `:`
* **Importance**: Medium
* **Valid values / notes**: Applied to non-collection targets; leading and trailing whitespace is removed from the configured value first.

#### `redis.charset`

Character set used to encode Redis key and value strings.

* **Type**: `string`
* **Default**: The character set returned by `Charset.defaultCharset().name()`
* **Importance**: High
* **Valid values / notes**: Must be a character-set name recognized by the JVM; conversion of Kafka record keys and values also depends on the Converter configuration.

### Write Delivery

#### `redis.multiexec`

Whether to use Redis MULTI/EXEC transactions for batch writes.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / notes**: Supports only `XADD`, `LPUSH`, `RPUSH`, `SADD`, and `ZADD`; it cannot be used with `HSET`, `JSONSET`, `TSADD`, `SET`, or `DEL`.

#### `redis.wait.replicas`

Number of Redis replicas to wait for confirmation after a write.

* **Type**: `int`
* **Default**: `0`
* **Importance**: Medium
* **Valid values / notes**: `0` means not to wait for replicas; values greater than `0` execute Redis WAIT and are subject to `redis.wait.timeout`.

#### `redis.wait.timeout`

Timeout for the Redis WAIT command, in milliseconds.

* **Type**: `long`
* **Default**: `1000`
* **Importance**: Medium
* **Valid values / notes**: Used with `redis.wait.replicas`; waiting occurs only when a replica count is configured.

### Kafka Consumption and Tasks

#### `topics`

List of Kafka topics consumed by this Sink.

* **Type**: `list`
* **Default**: `""`
* **Importance**: High
* **Valid values / notes**: Mutually exclusive with `topics.regex`; exactly one of the two must be non-empty.

#### `topics.regex`

Java regular expression used to match Kafka topics.

* **Type**: `string`
* **Default**: `""`
* **Importance**: High
* **Valid values / notes**: Mutually exclusive with `topics`; the expression must be a valid Java regular expression and must not match the DLQ topic.

#### `tasks.max`

Maximum number of Sink Tasks allowed to start.

* **Type**: `int`
* **Default**: `1`
* **Importance**: High
* **Valid values / notes**: Must be at least `1`; actual parallelism is also affected by the number of Kafka partitions and assignment results, and adding Tasks does not provide global ordering across partitions.

#### `key.converter`

Converter that deserializes Kafka message keys into Connect data.

* **Type**: `class`
* **Default**: None
* **Importance**: Low
* **Valid values / notes**: Use an instantiable Kafka Connect Converter; use `org.apache.kafka.connect.storage.StringConverter` or `org.apache.kafka.connect.converters.ByteArrayConverter` for string or byte keys, respectively.

#### `value.converter`

Converter that deserializes Kafka message values into Connect data.

* **Type**: `class`
* **Default**: None
* **Importance**: Low
* **Valid values / notes**: Use an instantiable Kafka Connect Converter; the value type required by the target command must match the Converter and SMT output.

### Error Handling

#### `errors.tolerance`

Controls whether Kafka Connect tolerates record-processing errors.

* **Type**: `string`
* **Default**: `none`
* **Importance**: Medium
* **Valid values / notes**: Can be `none` or `all`; this is a Kafka Connect error-handling setting and does not change the data-mapping requirements of Redis commands.

#### `errors.retry.timeout`

Total duration of Kafka Connect error retries, in milliseconds.

* **Type**: `long`
* **Default**: `0`
* **Importance**: Medium
* **Valid values / notes**: `0` means no Connect error retries, `-1` means unlimited retries, and other values specify the retry duration; this differs from the Redis command timeout configured by `redis.timeout`.

#### `errors.retry.delay.max.ms`

Maximum delay between Kafka Connect error retries, in milliseconds.

* **Type**: `long`
* **Default**: `60000`
* **Importance**: Medium
* **Valid values / notes**: Effective only when Connect error retries are enabled; the actual delay may include jitter.

#### `errors.deadletterqueue.topic.name`

Name of the DLQ topic used to store error records.

* **Type**: `string`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / notes**: An empty value disables the DLQ; the configured DLQ topic must not be consumed by `topics` or matched by `topics.regex`.

## Best Practices

### Write to a Redis String for the Initial Integration

Applicable business scenario: You need to first write string or byte values from Kafka to Redis, verify the topic, connection, and keyspace, and then gradually expand to other Redis data structures.

```properties theme={null}
connector.class=com.redis.kafka.connect.RedisSinkConnector
topics=orders
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
redis.host=<redis-host>
redis.port=6379
redis.command=SET
redis.key=${topic}
```

Key points: With `SET`, a non-empty record key participates in generating the target key; subsequent writes to the same target key overwrite the existing value. This configuration is the same as the Quick Start and is suitable as the minimum verification path for an initial integration.

### Write Structured Records to a Hash

Applicable business scenario: Kafka values have already been converted by an Avro or JSON Converter into a Struct or Map, and you need to write the fields to a Redis Hash for field-level reads.

```properties theme={null}
connector.class=com.redis.kafka.connect.RedisSinkConnector
topics=orders
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
redis.host=<redis-host>
redis.port=6379
redis.command=HSET
redis.key=${topic}
```

Key points: `HSET` requires a Struct or Map value, with field names used as Hash fields; a non-empty `redis.key` is combined with the record key to generate a non-collection target key. When using the JSON Converter, ensure that the input JSON structure can be converted to a Map; for schemaful data, use the corresponding structured Converter instead.

### Append Events to a Redis Stream

Applicable business scenario: You need to append Kafka events to a Redis Stream for continued processing by Redis Stream consumers, rather than retaining only the latest state for a key.

```properties theme={null}
connector.class=com.redis.kafka.connect.RedisSinkConnector
topics=orders
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
redis.host=<redis-host>
redis.port=6379
redis.command=XADD
redis.key=orders-stream
redis.multiexec=true
```

Key points: `XADD` uses the fields of a Struct or Map as the Stream message body; `redis.multiexec=true` applies only to the command set that supports MULTI/EXEC, which includes `XADD`. This mode appends writes, so at-least-once delivery or replay may create duplicate Stream messages; the task count, Kafka partitions, and restarts do not provide a global order across partitions, and this mode cannot serve as an exactly-once guarantee.

## Monitoring

### What to Monitor

Monitor Kafka Connect health, Connector and Task status, throughput, latency, offset commits, errors, retries, and Worker JVM signals; observe Redis connection errors, command timeouts, and batch-write latency together with Task errors and retries; monitor DLQ activity only when the corresponding error handling is enabled.

### Import the Grafana Dashboard

Confirm that Connect metrics are connected to a Grafana data source and that the collected labels satisfy 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 in Grafana, and select the corresponding data source.

## Limitations

* `redis.multiexec=true` supports only `XADD`, `LPUSH`, `RPUSH`, `SADD`, and `ZADD`, and cannot be used with the Redis Cluster client.
* The Connector does not provide a general Schema-to-Redis-data-structure mapping; `HSET` and `XADD` require Struct or Map values, while `ZADD` and `TSADD` require numeric input.
* The Connector uses at-least-once delivery semantics and does not provide exactly-once processing, Connector-level idempotency, or duplicate suppression; append commands such as List and Stream commands may append again during replay.
* The Connector has no native batch-size configuration; each set of records delivered by Kafka Connect to a Task is processed as one Writer batch, and an oversized Worker batch may be affected by Redis timeouts and server capacity.
* `topics` and `topics.regex` are mutually exclusive, and the Sink must configure exactly one of them as non-empty; the DLQ topic cannot also be consumed by this Sink.

## FAQ

### The Task Cannot Connect to Redis After Starting

Check whether `redis.uri` is empty and whether `redis.host` and `redis.port` are correct; a non-empty URI takes precedence over the host and port. When TLS is enabled, check that `redis.tls`, the CA file, client certificate, and private-key path are readable by the Worker; when ACL is enabled, check that `redis.username` and `redis.password` match. After correcting the connection configuration, confirm that the Task resumes running and observe whether Redis receives the connection.

### A Value-Type Error Occurs When Writing to a Hash or Stream

`HSET` and `XADD` require Struct or Map values. Check `value.converter` and the SMT output to confirm that the record value is not an ordinary string or an unsupported object; the JSON Converter should output a structure that can be converted to a Map. If the business data is itself a string, use `SET` or `JSONSET` instead of relying on the Connector to parse the string automatically into Hash fields.

### The Redis Key Written for a Record Is Not as Expected

Check `redis.key`, `redis.separator`, and the Kafka record key. `${topic}` is replaced with the source topic; collection types use only the expanded `redis.key`, while non-collection types usually also append the record key. The record key must be convertible to a string or bytes; an empty or unsupported record key may cause conversion to fail before the write.

### Duplicate Events Appear in Redis After a Restart

This Sink uses at-least-once delivery. If a failure occurs after the target write completes but before the offset is persisted, the record may be replayed after a restart. `SET`, `HSET`, `JSONSET`, and `DEL` usually overwrite or delete the same target key, while `LPUSH`, `RPUSH`, and `XADD` may append again; when duplicates must be avoided, design idempotency keys or deduplication logic at the business layer.

### The Configuration Is Invalid After Enabling `redis.multiexec`

Confirm that `redis.command` is `XADD`, `LPUSH`, `RPUSH`, `SADD`, or `ZADD`, and confirm that `redis.cluster=true` is not set. Other commands are rejected by configuration validation; the Redis Cluster client also does not support this transaction path.