> ## 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 Stream Source Connector > Configure and operate the Redis Stream Source Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## 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](../manage-connectors). ```properties theme={null} connector.class=com.redis.kafka.connect.RedisStreamSourceConnector redis.host= redis.stream.name= key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter ``` Replace `` with the Redis address and `` 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` 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. ```properties theme={null} connector.class=com.redis.kafka.connect.RedisStreamSourceConnector redis.host= redis.port= redis.tls=true redis.cacert= redis.username= redis.password= redis.stream.name= key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter ``` **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. ```properties theme={null} connector.class=com.redis.kafka.connect.RedisStreamSourceConnector redis.host= redis.stream.name= batch.size=1000 redis.stream.block=500 key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter ``` **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. ```properties theme={null} connector.class=com.redis.kafka.connect.RedisStreamSourceConnector redis.host= redis.stream.name= tasks.max=2 redis.stream.consumer.name=consumer-${task} key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter ``` **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](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 * 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`.