Skip to main content

​
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.
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.
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.
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.
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, 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.