<a id="cc-redis-kafka-source"></a>

# Redis Kafka Source Connector for Confluent Cloud

The fully managed Redis Kafka Source connector for Confluent Cloud moves data from
a Redis database into an Apache Kafka® cluster. The connector captures changes
from Redis data structures (streams and keys) and publishes them to a Kafka
topic, enabling real-time data streaming from Redis to Kafka.

For example, when you configure the connector to monitor a Redis stream, it
reads stream entries and publishes messages to a Kafka topic. Similarly, when
monitoring Redis keyspace notifications, the connector captures changes to
keys in a Redis database, and publishes keys and values to a Kafka topic. The
connector maps the data structure key to the record key, and the value to the record value.

## Features

The connector offers the following features:

* **Delivery guarantees**: The Stream Source connector can be configured to acknowledge stream messages
  either automatically (for at-most-once delivery) or explicitly (for at-least-once delivery). The default
  is at-least-once delivery.

  The Keys Source connector does not guarantee data consistency because it relies on Redis keyspace
  notifications, which have no delivery guarantees. It is possible to miss some notifications,
  for example, due to network failures.
* **Database authentication:** The connector supports password-based authentication for all Redis services and IAM role-based
  authorization using Confluent Provider Integration for AWS MemoryDB and AWS ElastiCache.
  For more information about provider integration setup, see the [IAM roles authentication](#cc-redis-kafka-source-setup-connection).
* **Supported data formats**: The connector supports AVRO, JSON_SR (JSON Schema), or PROTOBUF output
  formats. [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) must be enabled to use these Schema Registry-based formats.
* **Client-side encryption (CSFLE and CSPE) support**: The connector supports CSFLE or CSPE for sensitive data.
  For more information about CSFLE or CSPE setup, see the [connector configuration](#cc-redis-kafka-source-setup-connection).
* **Unified source connector experience**: The connector offers a unified experience by combining the
  functionalities of a Keys Source Connector and a Stream Source Connector. Set `source.type` to `KEYS` or `STREAM` when you set up the connector.
  - **KEYS**: Captures key changes in a Redis database and publishes keys and values to a Kafka topic. It maps the data
    structure key to the record key and the value to the record value. Ensure keyspace notifications are enabled using
    `notify-keyspace-events KEA` in `redis.conf` or by running `CONFIG SET notify-keyspace-events KEA`.
    For details, see [Redis Keyspace Notifications](https://redis.io/docs/latest/develop/pubsub/keyspace-notifications/).
  - **STREAM**: Reads from a Redis stream and publishes messages to a Kafka topic.
* **Supports multiple tasks**: The Stream Source connector supports running one or more tasks. More tasks may improve performance.

  The Keys Source connector can be configured with only one task.
* **Secret manager integration**: The connector supports secret manager integration. For `Password` based authentication, the connector can retrieve the following configurations from an integrated secret manager at runtime as needed.

  | **Secret manager managed configuration**   | **Type**   |
  |--------------------------------------------|------------|
  | `redis.username`                           | `STRING`   |
  | `redis.password`                           | `PASSWORD` |

  For more information, see [Create a secret manager integration in Confluent Cloud](secret-manager-integration/overview.md#cloud-secret-manager-quickstart).

For more information and examples to use with the Confluent Cloud API for Connect,
see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section.

## Limitations

Be sure to review the following information.

* For connector limitations, see [Redis Kafka Source](limits.md#cc-redis-kafka-source-limits) limitations.
* If you plan to use one or more Single Message Transformations (SMTs), see [SMT Limitations](single-message-transforms.md#cc-single-message-transforms-limitations).

<a id="cc-redis-kafka-source-topic-sizing"></a>

## Maximum message size

This connector creates topics automatically. When it creates topics, the internal connector configuration property `max.message.bytes` is set to the following:

* Basic cluster: `8 MB`
* Standard cluster: `8 MB`
* Enterprise cluster: `8 MB`
* Dedicated cluster: `20 MB`

For more information about Confluent Cloud clusters, see [Kafka Cluster Types in Confluent Cloud](../clusters/cluster-types.md#cloud-cluster-types).

## Quick Start

Use this quick start to get up and running with the Confluent Cloud Redis Kafka
Source connector. The quick start provides the basics of selecting the connector
and configuring it to consume data from Redis database and persist the data to Kafka.

<a id="cc-redis-kafka-source-prereqs"></a>

Prerequisites
: - Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/) cluster
    on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud.
  - The Confluent CLI installed and configured for the cluster.
    For more information, see [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html).
  - [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format
    (for example, AVRO, JSON_SR (JSON Schema), or PROTOBUF).
  - Access to a Redis server.
  - For networking considerations, see [Networking and DNS](overview.md#connect-internet-access-resources).
    To use a set of public egress IP addresses, see [Public Egress IP Addresses for Confluent Cloud Connectors](static-egress-ip.md#cc-static-egress-ips).
  <br/>
  - Kafka cluster credentials. The following lists the different ways you can provide credentials.
    - Enter an existing [service account](service-account.md#s3-cloud-service-account) resource ID.
    - Create a Confluent Cloud [service account](service-account.md#s3-cloud-service-account) for the connector. Make sure to review the ACL entries required in the [service account documentation](service-account.md#s3-cloud-service-account). Some connectors have specific ACL requirements.
    - Create a Confluent Cloud API key and secret. To create a key and secret, you can use [confluent api-key create](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) *or* you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector.

### Using the Confluent Cloud Console

#### Step 1: Launch your Confluent Cloud cluster

To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../get-started/index.md#cloud-create-kafka-cluster).

#### Step 2: Add a connector

In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add
connector**.

#### Step 3: Select your connector

Click the **Redis Kafka Source** connector card.

![Redis Kafka Source Connector Card](images/ccloud-redis-kafka-source-icon.png)

<a id="cc-redis-kafka-source-setup-connection"></a>

#### Step 4: Enter the connector details

#### NOTE
* Make sure you have all your [prerequisites](#cc-redis-kafka-source-prereqs) completed.
* An asterisk ( \* ) designates a required entry.

At the **Redis Kafka Source Connector** screen, complete the following:

### Kafka access

1. Select the way you want to provide **Kafka Cluster credentials**. You can
   choose one of the following options:
   - **My account**: This setting allows your connector to globally access everything
     that you have access to. With a user account, the connector uses an API key and
     secret to access the Kafka cluster. This option is not recommended for production.
   - **Service account**: This setting limits the access for your connector by using a
     [service account](service-account.md#s3-cloud-service-account). This option is recommended for
     production.
   - **Use an existing API key**: This setting allows you to specify an API key and a
     secret pair. You can use an existing pair or create a new one. This method is not
     recommended for production environments.

   #### NOTE
   Freight clusters support only service accounts for Kafka authentication.
2. Click **Continue**.

### Authentication

1. Configure the authentication properties:

   **Authentication method**
   - **Service name**: Specifies the Redis service to use.
     * If you select **AWS ElastiCache**, the connector connects to Amazon ElastiCache.
     * If you select **AWS MemoryDB**, the connector connects to Amazon MemoryDB.
     * If you select **Others**, the connector connects to any other Redis service (such as self-hosted or Redis Cloud).

     Note that IAM role authentication is supported only for `AWS ElastiCache` and `AWS MemoryDB`. For other services, use password authentication.
   - **Authentication method**: Select how you want to authenticate with Redis.
     * If you select **Password**, enter your **Redis password**.
     * If you select **IAM Roles**, choose an existing integration name under Provider integration name dropdown that has access to your resource. For more information, see [Manage Provider Integration for Fully Managed Connectors in Confluent Cloud](provider-integration.md#cloud-pi-quickstart).
   - **Use secret manager**: Fetch sensitive configuration values from a secret manager.

   **Secret manager configuration**
   - **Secret manager**: Select the secret manager to use for retrieving sensitive data.
   - **Configurations from Secret manager**: Select the configurations whose values to fetch from the secret manager.
   - **Provider Integration**: Select an existing integration that has access to your resource if you select **IAM Roles** as your authentication method.

   **Connection credentials**
   - **Redis hostname**: The IP address or hostname of the Redis database server.
   - **Redis port number**: The port number used to connect to Redis database server.
   - **Redis database index**: The numerical index of the Redis database you want to write data to.
   - **Redis username**: The username of the Redis user connecting to the Redis database server.
   - **Redis server mode**: Specify whether the Redis server runs on a single node or
     across multiple nodes. The default value is `Standalone`.

     #### NOTE
     You must configure this setting to match your actual Redis deployment.
     If you select `Standalone` mode but connect to a Redis Cluster, the connector will
     use a single-node client and may throw `CROSSSLOT` exceptions when handling multi-key
     operations. Always set this to `Cluster` when connecting to a Redis Cluster deployment.
   - **Redis password**: The password of the Redis user connecting to the Redis database server.
   - **Provider Integration**: Select an existing integration that has access to your resource if you select **IAM Roles** as your authentication method.
   - **Cluster name**: The name of the `AWS ElastiCache` or `AWS MemoryDB` cluster to connect to. This field is required when **Service name** is set to **AWS ElastiCache** or **AWS MemoryDB** and **Authentication method** is set to **IAM Roles**.
   - **ElastiCache serverless mode**: Specify whether the connection target is an ElastiCache serverless configuration. This setting affects how the IAM authentication token is generated for serverless endpoints.
     * Set to `true` if you are connecting to an ElastiCache serverless endpoint.
     * Set to `false` (default) for standard ElastiCache clusters.

     This field is only applicable when **Service name** is set to **AWS ElastiCache** and **Authentication method** is set to **IAM Roles**.
   - **AWS Region**: The AWS region where the cluster is deployed (for example, `us-east-1` or `eu-west-1`). This must match the region of the ElastiCache or MemoryDB cluster for IAM authentication to work correctly. This field is required when **Service name** is set to **AWS ElastiCache** or **AWS MemoryDB** and **Authentication method** is set to **IAM Roles**.

   **TLS configuration**
   - **Enable TLS**: Specify whether to use Transport Layer Security (TLS) to connect
     to the Redis database. Defaults to `false`.
   - **CA certificate file**: If you set `Enable TLS` property to `true`, upload the CA
     certificate file. The connector uses a TLS-encrypted connection and verifies the server’s
     TLS certificate against the configured Certificate Authority (CA) certificates. You can
     use this with or without client certificates.

   **Client certificate authentication**
   - **Private key file**: Upload the private key file in PEM format. Use this file along
     with the certificate file for client certificate authentication.
   - **Certificate file**: Upload the X.509 certificate chain file in PEM format. Use this file along
     with the private key file for client certificate authentication.
   - **Key password**: Specify the password of the private key file if it is password-protected.
2. Click **Continue**.

### Configuration

Configure the following:

- **Source type**: Defines the type of Redis Kafka source connector to use. Select `KEYS` to monitor Redis
  keyspace notifications, or `STREAM` to read from Redis Streams.
- **Output Kafka record value format**: Sets the output Kafka record value
  format (data going to the Kafka topic). Valid entries are AVRO, JSON,
  JSON_SR (JSON Schema), PROTOBUF, STRING, or BYTES. Note that you
  need to have [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) configured if using
  a schema-based message format like AVRO, JSON_SR, and PROTOBUF.

**Keys source settings**

- **Keys pattern**: Specifies the keyspace glob-style pattern to subscribe to. Use `*` to subscribe to all keys.
- **Key reading mode**: Controls the data reading behavior. Use `LIVE` for both the initial snapshot and subsequent updates, or `LIVEONLY` to process only updates.
- **Idle timeout**: Sets the idle timeout in milliseconds. Use `0` to disable.
- **Batch size**: Controls the maximum size of the batch for writing into a topic. Defaults to `100`.

**Stream source settings**

- **Stream name**: Specifies the name of the Redis stream to read from.
- **Stream Offset**: Determines the stream offset to start reading from. Use `0-0` to start from the
  beginning, `$` to read only new messages, or a specific offset, for example `1234567890-0`.
- **Block timeout**: Sets the maximum amount of time (in milliseconds) to wait while polling for
  stream messages (`XREAD [BLOCK milliseconds]`).
- **Delivery guarantee**: Specifies the stream message delivery guarantee. Options are `at-least-once` or `at-most-once`.
- **Consumer group**: Specifies the group name of the stream consumer. The connector creates this group if it does not exist.
- **Consumer name**: Specifies a format string for the stream consumer, which may contain `${task}` as
  a placeholder for the task ID. For example, `consumer-${task}` for a task ID `123` will
  map to the consumer name `consumer-123`.
- **Batch size**: Controls the maximum size of the batch for writing into a topic. Defaults to `100`.

**Data encryption**

- Enable **Client-Side Field Level Encryption**
  for data encryption. Specify a **Service Account** to
  access the Schema Registry and associated encryption rules or keys with that schema. For more
  information on CSFLE or CSPE setup,
  see [Manage encryption for connectors](csfle.md#connect-csfle).

### **Show advanced configurations**

- **Schema context**: Select a schema context to use for this connector, if using
  a schema-based data format. This property defaults to the **Default** context,
  which configures the connector to use the default schema set up for Schema Registry in your
  Confluent Cloud environment. A schema context allows you to use separate schemas (like
  schema sub-registries) tied to topics in different Kafka clusters that share the
  same Schema Registry environment. For example, if you select a non-default context, a
  **Source** connector uses only that schema context to register a schema and a
  **Sink** connector uses only that schema context to read from. For more
  information about setting up a schema context, see [What are schema contexts and when should you use them?](../sr/faqs-cc.md#faq-schema-contexts).
- **Command timeout**: Set the Redis command timeout in seconds. Defaults to `60`.
- **Connection pool size**: Specify the maximum number of connections in the pool. Defaults to `8`.

**Additional Configs**

- **Value Converter Replace Null With Default**: Specifies whether to replace fields that have a default value and that are null to the default value. When set to `true`, the connector uses the default value; otherwise, it uses `null`. Applies to the `JSON` converter.
- **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`.
- **Value Converter Schemas Enable**: Includes schema within each of the serialized values. Input messages must contain `schema` and `payload` fields and must not contain additional fields. For plain `JSON` data, set this to `false`. Applies to the `JSON` converter.
- **Errors Tolerance**: Use this property to configure the connector’s error handling behavior.

  #### WARNING
  Use this property with caution for sink connectors, as it can lead to data loss. If you set this property to `all`, the connector does not fail on errant records, but logs them (and sends to DLQ for sink connectors) and continues processing. If you set this property to `none`, the connector task fails on errant records.
- **Value Converter Ignore Default For Nullables**: When set to `true`, this property ensures that the corresponding record in Kafka is `null`, instead of showing the default column value. Applies to the `AVRO`, `PROTOBUF`, and `JSON_SR` converters.
- **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals:
  `BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and
  `NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value.
- **Key Converter Schema ID Serializer**: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
- **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters.
- **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry.
- **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration.
- **Value Converter Schema ID Serializer**: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.

**Auto-restart policy**

- **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its
  task in the event of user-actionable errors. Defaults to `true`, enabling the connector to
  automatically restart in case of user-actionable errors. Set this property to `false` to
  disable auto-restart for failed connectors. If disabled, you must manually restart the connector.

**Transforms**

- **Single Message Transformations**: To add a new SMT, see [Add transforms](single-message-transforms.md#cc-single-message-transforms-ui).
  For more information about unsupported SMTs, see
  [Unsupported transformations](single-message-transforms.md#cc-single-message-transforms-unsupported-transforms).

**Processing position**

- **Set offsets**: Click **Set offsets** to define a specific offset for
  this connector to begin procession data from. For more information
  on managing offsets, see [Manage offsets](offsets.md#connect-custom-offsets).

For all property values and definitions, see [Configuration Properties](#cc-redis-kafka-source-config-properties).

- Click **Continue**.

### Sizing

Based on the number of topic partitions you select, you will be provided
with a recommended number of tasks.

1. To change the number of recommended tasks, enter the number of
   [tasks](/platform/current/connect/concepts.html#tasks) for the connector to use
   in the **Tasks** field.
2. Click **Continue**.

### Review and Launch

1. Verify the connection details by previewing the running configuration.
2. After you’ve validated that the properties are configured to your
   satisfaction, click **Launch**.

   The status for the connector should go from **Provisioning** to
   **Running**.

#### Step 5: Check the Kafka topic

After the connector is running, verify that Redis records are populating the
Kafka topic.

For more information and examples to use with the Confluent Cloud API for Connect,
see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section.

### Using the Confluent CLI

Complete the following steps to set up and run the connector using the Confluent CLI.

#### NOTE
Make sure you have all your [prerequisites](#cc-redis-kafka-source-prereqs) completed.

#### Step 1: List the available connectors

Enter the following command to list available connectors:

```none
confluent connect plugin list
```

#### Step 2: List the connector configuration properties

Enter the following command to show the connector configuration properties:

```none
confluent connect plugin describe <connector-plugin-name>
```

The command output shows the required and optional configuration properties.

#### Step 3: Create the connector configuration file

Create a JSON file that contains the connector configuration properties. The
following example shows the required connector properties.

```json
{
   "connector.class": "RedisKafkaSource",
   "name": "<my-connector-name>",
   "kafka.auth.mode": "KAFKA_API_KEY",
   "kafka.api.key": "<my-kafka-api-key>",
   "kafka.api.secret": "<my-kafka-api-secret>",
   "kafka.topic": "keys_source",
   "redis.host": "test-18211.c90.us-east-1.ec2.redis-cloud.com",
   "redis.port": "18211",
   "redis.database": "0",
   "redis.username": "default",
   "redis.password": "********************",
   "redis.tls": "false",
   "source.type": "KEYS",
   "keys.batch.size": "100",
   "redis.timeout": "60",
   "redis.pool": "8",
   "redis.server.mode": "Standalone",
   "redis.keys.pattern": "*",
   "mode": "LIVE",
   "redis.keys.timeout": "0",
   "output.data.format": "AVRO",
   "tasks.max": "1",
   "auto.restart.on.user.error": "true"
}
```

Note the following property definitions:

* `"connector.class"`: Identifies the connector plugin name.
* `"name"`: Sets a name for your new connector.

* `"kafka.auth.mode"`: Identifies the connector authentication mode you want to use. There are two options: `SERVICE_ACCOUNT` or `KAFKA_API_KEY` (the default). To use an API key and secret, specify the configuration properties `kafka.api.key` and `kafka.api.secret`, as shown in the example configuration (above).  To use a [service account](service-account.md#s3-cloud-service-account), specify the **Resource ID** in the property `kafka.service.account.id=<service-account-resource-ID>`. To list the available service account resource IDs, use the following command:
  ```bash
  confluent iam service-account list
  ```

  For example:
  ```bash
  confluent iam service-account list

     Id     | Resource ID |       Name        |    Description
  +---------+-------------+-------------------+-------------------
     123456 | sa-l1r23m   | sa-1              | Service account 1
     789101 | sa-l4d56p   | sa-2              | Service account 2
  ```

* `"redis.host"`: The IP address or hostname of the Redis database server.
* `"redis.port"`: The port number used to connect to Redis database server.
* `"redis.tls"`: Specify whether to use Transport Layer Security (TLS) to connect
  to the Redis database. Defaults to `false`.
* `"source.type"`: Defines the type of Redis Kafka source connector to use. Select `KEYS` to monitor Redis
  keyspace notifications, or `STREAM` to read from Redis Streams.
* `"kafka.topic"`: Specifies the name of the destination Kafka topic where the connector
  publishes Redis events.
* `"keys.batch.size"`: Controls the maximum size of the batch for writing into a topic. Defaults to `100`.
  For the Stream Source connector, the property name is `"stream.batch.size"`.
* `"output.data.format"`: Sets the output Kafka record value format (data
  coming from the connector). Valid entries are `AVRO`, `JSON_SR`, or
  `PROTOBUF`. You must have Confluent Cloud Schema Registry
  configured to use these schema-based message formats.
* `"tasks.max"`: Enter the maximum number of
  [tasks](/platform/current/connect/index.html#tasks) for the connector to use. More
  tasks might improve performance.

  #### NOTE
  The Keys Source connector can be configured with only one task.

#### NOTE
To enable CSFLE or CSPE for data encryption, specify the following properties:

* `csfle.enabled`: Flag to indicate whether the connector honors CSFLE or CSPE rules.
* `sr.service.account.id`: A Service Account to access the Schema Registry and associated encryption rules or keys with that schema.

For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](csfle.md#connect-csfle).

**SMTs**: For details about adding SMTs using the Confluent CLI, see the [Single Message Transformations](single-message-transforms.md#cc-single-message-transforms) documentation.

See [Configuration Properties](#cc-redis-kafka-source-config-properties) for all property values and
definitions.

#### Step 4: Load the properties file and create the connector

Enter the following command to load the configuration and start the connector:

```none
confluent connect cluster create --config-file <file-name>.json
```

For example:

```none
confluent connect cluster create --config-file redis-kafka-source.json
```

Example output:

```none
Created connector confluent-redis-kafka-source lcc-ix4dl
```

#### Step 5: Check the connector status

Enter the following command to check the connector status:

```none
confluent connect cluster list
```

Example output:

```none
ID          |            Name           | Status  | Type
+-----------+---------------------------+---------+-------+
lcc-ix4dl   | confluent-redis-kafka-source  | RUNNING | source
```

#### Step 6: Check the Kafka topic.

After the connector is running, verify that Redis documents are populating the
Kafka topic.

For more information and examples to use with the Confluent Cloud API for Connect,
see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section.

<a id="cc-redis-kafka-source-config-properties"></a>

## Configuration Properties

Use the following configuration properties with the fully managed connector. For
self-managed connector property definitions and other details, see the connector
docs in [Self-managed connectors for Confluent Platform](/platform/current/connect/kafka_connectors.html).

### How should we connect to your data?

`name`
: Sets a name for your connector.
  <br/>
  * Type: string
  * Valid Values: A string at most 64 characters long
  * Importance: high

### Which topic do you want to send data to?

`kafka.topic`
: Identifies the topic name to write the data to.
  <br/>
  * Type: string
  * Importance: high

### Schema Config

`schema.context.name`
: Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.
  <br/>
  * Type: string
  * Default: default
  * Importance: medium

### Kafka Cluster credentials

`kafka.auth.mode`
: Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
  <br/>
  * Type: string
  * Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY
  * Importance: high

`kafka.api.key`
: Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
  <br/>
  * Type: password
  * Importance: high

`kafka.service.account.id`
: The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
  <br/>
  * Type: string
  * Importance: high

`kafka.api.secret`
: Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
  <br/>
  * Type: password
  * Importance: high

### Authentication method

`rediskafka.cluster.service.name`
: Specify the Redis service you are using. Choose AWS ElastiCache for Amazon ElastiCache, AWS MemoryDB for Amazon MemoryDB, or Others for any other Redis service (self-hosted, Redis Cloud, etc.). Note: IAM roles authentication is only supported for AWS ElastiCache and AWS MemoryDB services. For other services, use password authentication.
  <br/>
  * Type: string
  * Default: Others
  * Importance: medium

`authentication.method`
: Select how you want to authenticate the DB. Username Password or AWS IAM Roles.
  <br/>
  * Type: string
  * Default: Password
  * Importance: high

`secret.manager.enabled`
: Fetch sensitive configuration values from a secret manager.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: high

### Secret manager configuration

`secret.manager`
: Select the secret manager to use for retrieving sensitive data.
  <br/>
  * Type: string
  * Importance: high

`secret.manager.managed.configs`
: Select the configurations to fetch their values from the secret manager.
  <br/>
  * Type: list
  * Importance: high

`secret.manager.provider.integration.id`
: Select an existing provider integration that has access to your secret manager.
  <br/>
  * Type: string
  * Importance: high

### Redis connection

`redis.host`
: The hostname of Redis server to connect to.
  <br/>
  * Type: string
  * Importance: high

`redis.port`
: The port number of Redis server to connect to.
  <br/>
  * Type: string
  * Importance: high

`redis.database`
: The database index to write to.
  <br/>
  * Type: int
  * Default: 0
  * Importance: medium

`redis.username`
: The username of the Redis user connecting to the Redis database server.
  <br/>
  * Type: string
  * Importance: medium

`redis.server.mode`
: Whether redis server is running on one or multiple nodes.
  <br/>
  * Type: string
  * Default: Standalone
  * Importance: medium

`redis.password`
: The password of the Redis user connecting to the Redis database server.
  <br/>
  * Type: password
  * Importance: medium

`provider.integration.id`
: Select an existing integration that has access to your resource. In case you need to integrate a new IAM role, use provider integration
  <br/>
  * Type: string
  * Importance: high

`rediskafka.aws.credentials.provider.cluster.name`
: The name of your AWS ElastiCache or MemoryDB cluster that you want to connect to.
  <br/>
  * Type: string
  * Importance: medium

`rediskafka.aws.credentials.provider.elasticache.is.serverless`
: Set to true if you are connecting to an ElastiCache serverless configuration. This setting affects how the IAM authentication token is generated for ElastiCache serverless endpoints.
  <br/>
  * Type: boolean
  * Importance: medium

`rediskafka.aws.credentials.provider.cluster.region`
: The AWS region where your cluster is deployed (e.g., us-east-1, eu-west-1). This must match the region of your ElastiCache or MemoryDB cluster for IAM authentication to work correctly.
  <br/>
  * Type: string
  * Default: us-west-2
  * Importance: medium

### Redis security

`redis.tls`
: Establish a secure TLS connection to Redis.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: medium

`redis.cacert`
: X.509 CA certificate file to verify with. Use this with or without client certificates.
  <br/>
  * Type: password
  * Importance: medium

### Redis client certificate auth

`redis.key.file`
: Private key file (PEM format) to authenticate with. Use this file along with the certificate file for client certificate authentication.
  <br/>
  * Type: password
  * Importance: medium

`redis.key.cert`
: X.509 certificate chain file (PEM format) to authenticate with. Use this file along with the private key file for client certificate authentication.
  <br/>
  * Type: password
  * Importance: medium

`redis.key.password`
: Password of the private key file. Leave empty if key file is not password-protected.
  <br/>
  * Type: password
  * Importance: medium

### Source configuration

`source.type`
: Type of Redis source connector. Select `KEYS` to monitor Redis keyspace notifications, or `STREAM` to read from Redis Streams.
  <br/>
  * Type: string
  * Default: KEYS
  * Importance: high

`keys.batch.size`
: Number of records to process in each batch for writing into a topic.
  <br/>
  * Type: int
  * Default: 100
  * Valid Values: [1,…,10000]
  * Importance: medium

`stream.batch.size`
: Number of records to process in each batch for writing into a topic.
  <br/>
  * Type: int
  * Default: 100
  * Valid Values: [1,…,10000]
  * Importance: medium

`redis.timeout`
: Redis command timeout in seconds.
  <br/>
  * Type: long
  * Default: 60
  * Valid Values: [1,…,3600]
  * Importance: medium

`redis.pool`
: Maximum number of connections in the pool in the range of 1 to 100.
  <br/>
  * Type: int
  * Default: 8
  * Valid Values: [1,…,100]
  * Importance: medium

### Output messages

`output.data.format`
: Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR or PROTOBUF. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF
  <br/>
  * Type: string
  * Default: JSON_SR
  * Importance: high

### Keys source configuration

`redis.keys.pattern`
: Keyspace glob-style pattern to subscribe to. Use `*` to subscribe to all keys.
  <br/>
  * Type: string
  * Default: \*
  * Importance: medium

`mode`
: Use `LIVE` for snapshot + updates, `LIVEONLY` for just updates.
  <br/>
  * Type: string
  * Default: LIVE
  * Importance: high

`redis.keys.timeout`
: Idle timeout in milliseconds. Use `0` to disable.
  <br/>
  * Type: long
  * Default: 0
  * Valid Values: [0,…,3600000]
  * Importance: low

### Stream source configuration

`redis.stream.name`
: Name of the Redis stream to read from.
  <br/>
  * Type: string
  * Importance: high

`redis.stream.offset`
: Stream offset to start reading from. Use `0-0` to start from the beginning, `$` to read only new messages, or a specific offset like `1234567890-0`.
  <br/>
  * Type: string
  * Default: 0-0
  * Importance: medium

`redis.stream.block`
: The maximum amount of time in milliseconds to wait while polling for stream messages (XREAD [BLOCK milliseconds]).
  <br/>
  * Type: long
  * Default: 100
  * Valid Values: [0,…,60000]
  * Importance: low

`redis.stream.delivery`
: Stream message delivery guarantee. The valid options are `at-least-once` or `at-most-once`.
  <br/>
  * Type: string
  * Default: at-least-once
  * Importance: medium

`redis.stream.consumer.group`
: Stream consumer group name. This group will be created if it doesn’t exist.
  <br/>
  * Type: string
  * Default: kafka-consumer-group
  * Importance: high

`redis.stream.consumer.name`
: A format string for the stream consumer, which may contain ‘${task}’ as a placeholder for the task id. For example, ‘consumer-${task}’ for the task id ‘123’ will map to the consumer name ‘consumer-123’.
  <br/>
  * Type: string
  * Default: consumer-${task}
  * Importance: medium

### Number of tasks for this connector

`tasks.max`
: Maximum number of tasks for the connector.
  <br/>
  * Type: int
  * Default: 1
  * Importance: high

### Additional Configs

`header.converter`
: The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
  <br/>
  * Type: string
  * Importance: low

`producer.override.compression.type`
: The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.
  <br/>
  * Type: string
  * Importance: low

`producer.override.linger.ms`
: The producer groups together any records that arrive in between request transmissions into a single batched request. More details can be found in the documentation: [https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms](https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms).
  <br/>
  * Type: long
  * Valid Values: [100,…,1000]
  * Importance: low

`value.converter.allow.optional.map.keys`
: Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.auto.register.schemas`
: Specify if the Serializer should attempt to register the Schema.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.connect.meta.data`
: Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.enhanced.avro.schema.support`
: Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.enhanced.protobuf.schema.support`
: Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.flatten.unions`
: Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.generate.index.for.unions`
: Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.generate.struct.for.nulls`
: Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.int.for.enums`
: Whether to represent enums as integers. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.latest.compatibility.strict`
: Verify latest subject version is backward compatible when use.latest.version is true.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.object.additional.properties`
: Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.optional.for.nullables`
: Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.optional.for.proto2`
: Whether proto2 optionals are supported. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.scrub.invalid.names`
: Whether to scrub invalid names by replacing invalid characters with valid characters. Applicable for Avro and Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.use.latest.version`
: Use latest version of schema in subject for serialization when auto.register.schemas is false.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.use.optional.for.nonrequired`
: Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.wrapper.for.nullables`
: Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.wrapper.for.raw.primitives`
: Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`errors.tolerance`
: Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.
  <br/>
  * Type: string
  * Default: none
  * Importance: low

`key.converter.key.schema.id.serializer`
: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
  <br/>
  * Type: string
  * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer
  * Importance: low

`key.converter.key.subject.name.strategy`
: How to construct the subject name for key schema registration.
  <br/>
  * Type: string
  * Default: TopicNameStrategy
  * Importance: low

`value.converter.decimal.format`
: Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
  <br/>
  BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
  <br/>
  NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
  <br/>
  * Type: string
  * Default: BASE64
  * Importance: low

`value.converter.flatten.singleton.unions`
: Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`value.converter.ignore.default.for.nullables`
: When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`value.converter.reference.subject.name.strategy`
: Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
  <br/>
  * Type: string
  * Default: DefaultReferenceSubjectNameStrategy
  * Importance: low

`value.converter.replace.null.with.default`
: Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: low

`value.converter.schemas.enable`
: Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`value.converter.value.schema.id.serializer`
: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
  <br/>
  * Type: string
  * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer
  * Importance: low

`value.converter.value.subject.name.strategy`
: Determines how to construct the subject name under which the value schema is registered with Schema Registry.
  <br/>
  * Type: string
  * Default: TopicNameStrategy
  * Importance: low

### Auto-restart policy

`auto.restart.on.user.error`
: Enable connector to automatically restart on user-actionable errors.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: medium

<a id="cc-redis-kafka-source-faq"></a>

## Frequently asked questions

Find answers to frequently asked questions about the Redis Kafka Source connector for Confluent Cloud.

### Availability and deployment model

#### Can I run the Redis Kafka Source connector on a self-managed Connect cluster?

The Redis Kafka Source connector is available only as a fully managed connector
on Confluent Cloud. However, as a partner connector, it can also run on self-managed Kafka Connect clusters, though some features
like IAM role-based authentication are Confluent Cloud-only. For more information,
see the [Redis Kafka Connector on Confluent Marketplace](https://www.confluent.io/hub/redis/redis-kafka-connect).

### Authentication and connectivity

#### Which authentication methods are supported for Redis?

The connector supports the following authentication methods:

* **Password-based authentication:** Use `redis.password` for all Redis-compatible services including Redis Enterprise, ElastiCache, and MemoryDB.
* **IAM role-based authorization:** Available for AWS MemoryDB and AWS ElastiCache using Confluent Provider Integration. For more information, see [IAM roles authentication](cc-redis-kafka-db-sink.md#cc-redis-kafka-sink-setup-connection).

#### Why do I get `Could not authenticate to redis instance` errors?

Authentication failures typically occur due to one of the following reasons:

* **Incorrect credentials:** Verify that `redis.username` and `redis.password` are correct for your Redis instance.
* **IAM role misconfiguration:** If using AWS MemoryDB or ElastiCache with IAM authentication, ensure:
  - The Confluent Provider Integration is correctly configured.
  - The IAM role has the necessary permissions (`memorydb:Connect` or `elasticache:Connect`).
  - The `redis.username` matches the IAM user name.
* **TLS/SSL settings:** If your Redis instance requires TLS, ensure `redis.tls` is set to `true`.

**Resolution:**

1. Verify credentials are correct and match your Redis authentication method.
2. For IAM authentication, confirm Provider Integration setup using [IAM roles authentication](#cc-redis-kafka-source-setup-connection).
3. Test connectivity from Confluent Cloud to your Redis endpoint.
4. Check Redis server logs for authentication error details.

#### Why does the connector fail with connection timeout errors?

Connection timeouts point to networking or firewall issues:

* **Network accessibility:** Ensure your Redis instance is accessible from Confluent Cloud.
* **Firewall rules:** Verify that Confluent egress IP addresses are allowlisted in your Redis security groups or firewall.
* **VPC configuration:** If using VPC peering, confirm the network path is correctly configured.
* **Timeout settings:** The default timeout is `60` seconds. For slow networks, consider increasing `redis.timeout`.

**Checklist:**

1. Verify network connectivity from Confluent Cloud to Redis host and port.
2. Allowlist Confluent egress IPs or configure PrivateLink. See [Networking and DNS](overview.md#connect-internet-access-resources).
3. Ensure port (default `6379` or `6380` for SSL) is open.
4. For VPC peered clusters, verify VPC configuration.

### Configuration and setup

#### Why does the Keys Source connector not capture any data even though Redis keys are changing?

The Keys Source connector relies on Redis keyspace notifications, which must be explicitly enabled:

**Cause:** Redis keyspace notifications are disabled by default and must be configured on your Redis server.

**Resolution:**

1. Enable keyspace notifications by running `CONFIG SET notify-keyspace-events KEA` or configure `notify-keyspace-events KEA` in your `redis.conf` file.
2. Verify notifications are enabled: `CONFIG GET notify-keyspace-events`.
3. Ensure `redis.keys.pattern` matches your key pattern (default is `*` for all keys).

For more information, see [Redis Keyspace Notifications](https://redis.io/docs/latest/develop/pubsub/keyspace-notifications/).

#### What is the difference between `KEYS` and `STREAM` source types?

The connector offers two source types configured via the `source.type` property:

* **KEYS** (`source.type=KEYS`):
  - Captures key changes in a Redis database using keyspace notifications.
  - Maps Redis key to Kafka record key, Redis value to Kafka record value.
  - Single task only (`tasks.max=1`).
  - No delivery guarantees (relies on Redis keyspace notifications).
  - Requires keyspace notifications enabled: `notify-keyspace-events KEA`.
* **STREAM** (`source.type=STREAM`):
  - Reads from Redis Streams.
  - Supports multiple tasks for improved performance.
  - Configurable delivery guarantees (at-least-once by default, at-most-once if acknowledgment is disabled).
  - More reliable for production workloads requiring guaranteed delivery.

### Performance and tasks

#### Why is my Stream Source connector not consuming data fast enough?

Performance issues can result from several factors:

* **Insufficient tasks:** Increase `tasks.max` to run multiple tasks in parallel. More tasks may
  improve throughput for Stream Source connectors.
* **Batch size:** Adjust `stream.batch.size` (default `100`). Larger batches reduce overhead
  but increase latency.
* **Network latency:** High latency between Confluent Cloud and Redis can slow consumption. Consider
  deploying Redis closer to your Confluent Cloud cluster region.
* **Redis server load:** Check Redis server performance metrics (CPU, memory, network).

**Recommendations:**

1. For Stream Source connectors, try increasing `tasks.max`.
2. Adjust `stream.batch.size` based on your throughput requirements (100-1000).
3. Monitor connector lag and task metrics in the Confluent Cloud console.

#### NOTE
The Keys Source connector supports only one task (`tasks.max=1`).

### Data consistency and delivery guarantees

#### Why am I seeing duplicate records from the Stream Source connector?

The Stream Source connector provides at-least-once delivery by default, which can result in duplicates
during connector restarts or failures.

#### IMPORTANT
The Keys Source connector does not provide delivery guarantees and may miss events
due to the nature of Redis keyspace notifications.

## Next Steps

For an example that shows fully managed Confluent Cloud connectors in action with
Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html).
This example also shows how to use Confluent CLI to manage your resources in
Confluent Cloud.

[![image](images/topology.png)](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)