Skip to main content

​
Overview

The Redis Stream Source Connector reads messages from one Redis Stream, converts each Stream message into a Kafka Connect SourceRecord, and writes it to a Kafka topic. It sits between Redis and Kafka and is suitable for bringing business events, task statuses, or real-time notifications from a Redis Stream into a Kafka data pipeline. The Connector uses a Redis consumer group to distribute messages. The Kafka message key is the Redis Stream message ID, and the value is a structure containing id, stream, and the string field map body. The destination topic can be specified explicitly or generated from the Redis Stream name by using ${stream}.

​
License

Licensed under the Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka, and a Redis service, and confirm that the Connect Worker can access Redis and Kafka. If automatic topic creation is disabled in the Kafka cluster, create the destination topic before starting the connector. For preparation and management procedures, see Manage Connectors.
Replace <redis-host> with the Redis address and <redis-stream-name> with the Redis Stream key to read. The example uses the default Redis port 6379, the default consumer group, and the at-least-once delivery mode. By default, the destination topic has the same name as the Redis Stream. If authentication, TLS, or a fixed destination topic is required, add the corresponding configuration before creating the Connector.

​
Configuration

​
Redis Connection

​
redis.cluster

Selects the Redis standalone client or Redis Cluster client.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: false uses the standalone client, and true uses the Cluster client. The address or URI must match the selected deployment mode.

​
redis.host

The Redis hostname or address.
  • Type: string
  • Default: localhost
  • Importance: High
  • Valid Values / Notes: Used only when redis.uri is empty, together with redis.port. The address must be resolvable and accessible from the Connect Worker.

​
redis.port

The Redis TCP port.
  • Type: int
  • Default: 6379
  • Importance: High
  • Valid Values / Notes: Used only when redis.uri is empty, together with redis.host. ConfigDef does not validate the port range, so specify a valid port on which Redis is actually listening.

​
redis.uri

Specifies the connection endpoint with a Redis URI.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid Values / Notes: A non-empty value takes precedence over redis.host and redis.port. It supports client-accepted redis:// and rediss:// URIs. A URI can contain credentials and must be handled as sensitive information. A non-empty redis.password applies separately configured credentials after endpoint selection.

​
redis.timeout

The Redis command timeout in seconds.
  • Type: long
  • Default: 60
  • Importance: Medium
  • Valid Values / Notes: Use a positive number appropriate for Redis response latency. ConfigDef does not validate the range, and an invalid duration can fail during client construction or at runtime.

​
Redis Authentication

​
redis.username

The Redis ACL username.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid Values / Notes: Used only when redis.password is non-empty. Leave it empty to use password-only authentication.

​
redis.password

The Redis password.
  • Type: password
  • Default: Empty string
  • Importance: Medium
  • Valid Values / Notes: A non-empty value enables separate credential configuration. When redis.username is also set, ACL username and password authentication is used; otherwise, password-only authentication is used. Supply this value through a controlled Secret or Config Provider.

​
TLS

​
redis.tls

Explicitly enables TLS for a Redis connection constructed from redis.host and redis.port.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: Set it to true to enable TLS. When redis.uri uses rediss://, the connection uses TLS even if this setting remains false; however, redis.insecure is read only when this configuration is explicitly set to true.

​
redis.insecure

Disables TLS peer certificate verification.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: Effective only when redis.tls=true. Do not enable it in production. If only rediss:// is configured without setting redis.tls=true, this value does not affect the certificate verification branch.

​
redis.cacert

Specifies the CA certificate file used to verify the Redis server certificate.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid Values / Notes: A non-empty value must point to an X.509 CA certificate file readable by the Connect Worker.

​
redis.key.file

Specifies the TLS client private key file.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid Values / Notes: A non-empty value enables the client Key Manager. The file must be a PKCS#8 PEM private key readable by the Connect Worker, and a matching redis.key.cert must also be provided.

​
redis.key.cert

Specifies the TLS client certificate chain file.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid Values / Notes: Read only when redis.key.file is non-empty. It must point to a PEM X.509 certificate chain that matches the private key and is readable by the Connect Worker.

​
redis.key.password

Specifies the password for the TLS client private key.
  • Type: password
  • Default: Empty string
  • Importance: Medium
  • Valid Values / Notes: Read only when redis.key.file is non-empty. Leave it empty for an unencrypted private key. For an encrypted private key, supply the password through a controlled Secret or Config Provider.

​
Stream Selection and Routing

​
redis.stream.name

Specifies the Redis Stream key to read.
  • Type: string
  • Default: None
  • Importance: High
  • Valid Values / Notes: One Connector instance reads only one Stream. A usable Redis key must be provided. Although an explicitly empty string passes ConfigDef string parsing, it cannot form a valid deployment.
  • Required: Yes

​
topic

Specifies the Kafka topic to which SourceRecords are written.
  • Type: string
  • Default: ${stream}
  • Importance: Medium
  • Valid Values / Notes: ${stream} is replaced with the Redis Stream name. A fixed topic name can also be specified. The final name must meet Kafka topic naming, existence or automatic creation, and authorization requirements. Other placeholders are not resolved.

​
redis.stream.offset

Specifies the initial Redis Stream offset to use when no Kafka Connect source offset has been stored.
  • Type: string
  • Default: 0-0
  • Importance: Medium
  • Valid Values / Notes: Use a message ID accepted by the Redis Stream reader. On restart, a stored Kafka Connect offset takes precedence over this value. An existing consumer group also does not reset its server-side cursor when this value changes.

​
Consumer Group and Delivery

​
redis.stream.consumer.group

Specifies the Redis Stream consumer group name.
  • Type: string
  • Default: kafka-consumer-group
  • Importance: Medium
  • Valid Values / Notes: All tasks for the same Connector use this consumer group. Changing the group name switches the Redis Pending Entry and acknowledgment state used by the Connector.

​
redis.stream.consumer.name

Specifies the Redis Stream consumer name template.
  • Type: string
  • Default: consumer-${task}
  • Importance: Medium
  • Valid Values / Notes: ${task} is replaced with the numeric task ID. When tasks.max is greater than 1, retain this placeholder to prevent multiple tasks from using the same consumer identity.

​
redis.stream.delivery

Selects the Redis Stream message acknowledgment mode.
  • Type: string
  • Default: at-least-once
  • Importance: Medium
  • Valid Values / Notes: Accepts only the case-sensitive values at-least-once or at-most-once. The former acknowledges messages when the SourceTask commits and can replay them. The latter acknowledges messages immediately after reading, so a failure window can cause message loss. Neither provides an exactly-once transaction across Redis and Kafka.

​
Read Cadence

​
batch.size

Sets the maximum number of records for each Redis read request.
  • Type: int
  • Default: 500
  • Importance: Low
  • Valid Values / Notes: Use a positive number. It limits the number of records in one read request, not the total number of Pending Entries or the total memory across multiple polls. ConfigDef does not validate the bounds.

​
redis.stream.block

Sets the blocking duration for Redis XREADGROUP, in milliseconds.
  • Type: long
  • Default: 100
  • Importance: Low
  • Valid Values / Notes: Use a non-negative value supported by the Redis client. This value directly affects how long one poll waits when no new message is available. ConfigDef does not validate the range.

​
Compatibility Configuration

​
redis.pool

The inherited Redis connection pool size configuration.
  • Type: int
  • Default: 8
  • Importance: Medium
  • Valid Values / Notes: This configuration is publicly settable, but the Stream read path in RedisStreamSourceConnector 0.9.1 does not use it. Changing it has no confirmed Stream Source tuning effect.

​
Kafka Connect Framework

​
connector.class

Specifies the Connector implementation class to load.
  • Type: string
  • Default: None
  • Importance: High
  • Valid Values / Notes: Must be set to com.redis.kafka.connect.RedisStreamSourceConnector.
  • Required: Yes

​
tasks.max

Sets the maximum number of tasks that Kafka Connect creates for the Connector.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid Values / Notes: Must be at least 1. Every task reads the same Redis Stream and consumer group. With multiple tasks, retain redis.stream.consumer.name with the ${task} placeholder.

​
tasks.max.enforce

Controls whether the framework enforces the tasks.max limit.
  • Type: boolean
  • Default: true
  • Importance: Low
  • Valid Values / Notes: Kafka Connect has deprecated this configuration and plans to remove it in a future major version. Keep the default and use tasks.max to set the scale.
  • Deprecated: Yes
  • Replacement: None

​
key.converter

Specifies the Kafka Connect Converter for the Redis Stream message ID key.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid Values / Notes: When omitted, inherits the Worker’s key Converter. The Connector emits a non-null key with a STRING schema, so select a Converter compatible with that schema.

​
value.converter

Specifies the Kafka Connect Converter for the structured message value.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid Values / Notes: When omitted, inherits the Worker’s value Converter. The Connector emits a STRUCT containing id, stream, and a MAP<STRING,STRING> type body, so select a Converter that supports these schemas.

​
Best Practices

​
Enable TLS and ACL Authentication for Production Onboarding

Applicable Scenario: Redis is on a production network or uses ACLs to manage access. The connection between the Connect Worker and Redis must be encrypted, while the Connector is granted only the permissions needed to read the target Stream and use the consumer group. Configuration Example: Add the following connection configuration to the Quick Start configuration.
Key Notes: The CA file must be readable on every Worker that can run a task, and the username and password should be injected through a controlled Secret or Config Provider. Do not set redis.insecure=true in production. If Redis requires mutual TLS, also configure a matching client private key and certificate chain.

​
Tune the Read Cadence for Throughput and Idle Latency

Applicable Scenario: The Connector is reading a single Stream reliably, and you need to balance per-request processing overhead, throughput, and return latency when no new messages are available. Start with the defaults, observe throughput, latency, and Worker resources, and then make small adjustments. Configuration Example: The following example increases the number of records per request and allows idle reads to wait longer.
Key Notes: batch.size is the record count limit for one Redis read request, not a total memory or Pending Entry limit. redis.stream.block directly increases the wait time for an idle poll. The example values are not universally optimal. Adjust them gradually based on message size, destination topic write capacity, task latency, and Worker memory.

​
Add Tasks When a Single Stream Has a Sustained Backlog

Applicable Scenario: A single task has become the read bottleneck, the Redis Stream consumer group has a sustained backlog, and Kafka and the Connect Workers still have processing capacity. Adding tasks allows multiple Redis consumers in the same consumer group to process new messages. Configuration Example: The following example creates two tasks for the same Stream and retains task-specific consumer names.
Key Notes: Multiple tasks do not assign different Streams to different tasks. They still read the same Stream and consumer group. In version 0.9.1, all tasks also share one Kafka Connect source-offset key, while Redis Pending Entries remain owned by individual consumer identities. Keep consumer names stable and unique, and validate restart, rebalance, and scale-down recovery before production use; steady-state distribution does not establish an independent recovery checkpoint for each task. Multiple tasks and a multi-partition Kafka topic also do not provide global ordering across tasks or partitions.

​
Monitoring

​
What to Monitor

Monitor general Kafka Connect health, Connector and task status, throughput, latency, offset commits, errors, retries, and Worker JVM signals. Also monitor the Redis consumer group backlog and Pending Entry changes. Monitor DLQ activity only when the corresponding error handling is enabled in the deployment.

​
Import the Grafana Dashboard

Confirm that Kafka Connect metrics are available through a Grafana data source and that the collected labels meet the dashboard filter requirements. Download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

​
Limitations

  • One Connector instance can read only one Redis Stream key. Stream lists and pattern matching are not supported.
  • An existing Redis consumer group does not reset its server-side cursor when redis.stream.offset changes. On restart, a stored Kafka Connect source offset also takes precedence over this configuration.
  • The Connector does not provide an exactly-once transaction or duplicate suppression across Redis acknowledgment and Kafka writes. The at-least-once mode can replay messages, while the at-most-once mode can lose a message after acknowledgment and before it is written to Kafka.
  • The Connector recovers only Pending Entries owned by the current consumer identity. It does not automatically use XCLAIM or XAUTOCLAIM to take over messages owned by other consumers.
  • In version 0.9.1, all tasks use the same empty source partition and therefore share one Kafka Connect source-offset key. Do not assume that each Redis consumer has an independent recovery checkpoint; multi-task failure recovery requires separate validation.
  • In version 0.9.1, a successful commit in at-least-once mode does not clear the message IDs accumulated by the task. Later commits resend historical XACK IDs to Redis, and the in-memory list grows for the lifetime of the task. Long-running, high-throughput workloads can therefore increase task memory use and Redis acknowledgment overhead; no safe threshold has been verified.
  • Ordering is limited to the result of one poll by a single task. Multiple tasks or a multi-partition Kafka topic do not provide global ordering.
  • The message value schema is fixed as a map of string fields. It does not preserve arbitrary binary field values or automatically infer or evolve field types.

​
FAQ

​
Why Does Reading Not Start from the New Position After I Change redis.stream.offset?

A stored Kafka Connect source offset takes precedence over redis.stream.offset, and the Connector does not reset the server-side cursor when it opens an existing Redis consumer group. First check the consumer group in use, the Kafka Connect offset, and the Redis group state. To establish an independent read baseline, use a new consumer group and handle Pending Entries from the original group carefully instead of changing only the initial offset.

​
Why Do Duplicate Messages Appear in Kafka?

The default at-least-once mode acknowledges Redis messages when Kafka Connect commits, and there is no cross-system transaction among the Redis read, Kafka write, and offset commit. If a failure occurs after the write but before acknowledgment or offset persistence, a message can be replayed after restart. The Kafka message key is the Redis Stream message ID, which downstream systems can use for idempotent processing or deduplication.

​
Why Are Pending Entries from an Old Consumer Not Processed After Reducing the Number of Tasks?

The Connector reads only Pending Entries owned by the current consumer name and does not automatically take over messages owned by other consumers. Check the owner of each Pending Entry in the Redis consumer group. During scaling, keep the consumer names generated by consumer-${task} stable, handle a consumer’s Pending Entries before removing it, and explicitly transfer ownership through Redis operational procedures when necessary.

​
Why Does the Message Timestamp Differ from the Time in the Redis Stream ID?

The SourceRecord timestamp comes from the Worker clock when the Connector converts the message, not from the time component encoded in the Redis Stream ID. When event time is required, write the business timestamp explicitly into a Stream message field and have downstream systems extract and use it from body.