## Redis Sink Connector

## Description

The Redis Sink Connector is used to write data from Kafka to a Redis cache.

### Important

This connector expects records from Kafka to have a key and value that are stored as bytes or a string. If your data is already in Kafka in the format that you want in Redis consider using the `org.apache.kafka.connect.converters.ByteArrayConverter` or the `org.apache.kafka.connect.storage.StringConverter` for this connector. Keep in this does not need to be configured in the worker properties and can be configured at the connector level. If your data is not sitting in Kafka in the format you wish to persist in Redis consider using a Single Message Transformation to convert the data to a byte or string representation before it is written to Redis.

### Note

This connector supports deletes. If the record stored in Kafka has a null value, this connector will send a delete with the corresponding key to Redis.


### Configuration

#### General


##### `redis.hosts`

The Redis hosts to connect to.

*Importance:* High
*Type:* List
*Default Value:* [localhost:6379]


##### `redis.database`

Redis database to connect to.

*Importance:* Medium
*Type:* Int
*Default Value:* 1 (re-set to 0 in Dockerfile)


##### `redis.password`

Password used to connect to Redis.

*Importance:* Medium
*Type:* Password

##### `redis.ssl.enabled`

Flag to determine if SSL is enabled.

*Importance:* Medium
*Type:* Boolean
*Default Value:* false



##### `redis.ssl.keystore.password`

The password for the SSL keystore.

*Importance:* Medium
*Type:* Password
*Default Value:* [hidden]



##### `redis.ssl.keystore.path`

The path to the SSL keystore.

*Importance:* Medium
*Type:* String


##### `redis.ssl.truststore.password`

The password for the SSL truststore.

*Importance:* Medium
*Type:* Password


##### `redis.ssl.truststore.path`

The path to the SSL truststore.

*Importance:* Medium
*Type:* String



##### `redis.auto.reconnect.enabled`

Flag to determine if the Redis client should automatically reconnect.

*Importance:* Low
*Type:* Boolean
*Default Value:* true


##### `redis.request.queue.size`

The maximum number of queued requests to Redis.

*Importance:* Low
*Type:* Int
*Default Value:* 2147483647


##### Local development

Build the network with the connector and redis instances inside by running 

```
awsume dev
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 086679231553.dkr.ecr.us-east-1.amazonaws.com
docker-compose up --build -d
```

When done tear down the environment `docker compose down`.

To check values which are streamed to Redis instance use `docker compose exec redis sh` and run `redis-cli -a sOmE_sEcUrE_pAsS` inside the contaner `keys *` or `get [key]` to see all or specific key

To check kafka-connect instanse run `docker compose exec kafka-connect-redis-sink bash` to get shell or `docker compose exec kafka-connect-redis-sink [command inside the container]`

##### Example

This configuration is used typically along with [distributed mode](http://docs.confluent.io/current/connect/concepts.html#distributed-workers).
Write the following json to `connector.json`, configure all of the required values, and use the command below to
post the configuration to one the distributed connect worker(s).

```json
{
    "connector.class": "com.github.jcustenborder.kafka.connect.redis.RedisSinkConnector",
    "tasks.max": "1",
    "topics": "stream.s3.test.ibolshakov.local.split.keyed",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "redis.hosts": "redis:6379",
        "redis.password": "${aws:dev/kafka-connect-redis-sink/credentials:REDIS_PASSWORD}",
        "redis.database": "0",
    "config.providers": "aws",
    "config.providers.aws.class": "io.lenses.connect.secrets.providers.AWSSecretProvider",
    "config.providers.aws.param.aws.auth.method": "default",
    "config.providers.aws.param.aws.access.key": "dummy-client-key",
    "config.providers.aws.param.aws.secret.key": "dummy-secret-key",
    "config.providers.aws.param.aws.region": "us-east-1"
}
```

Use docker + curl to post the configuration to one of the Kafka Connect Workers. Change `http://localhost:8083/` the the endpoint of one of your Kafka Connect worker(s).

Create a new instance.
```bash
docker compose exec kafka-connect-redis-sink 'curl -s -X POST -H 'Content-Type: application/json' --data /tmp/config.json http://localhost:8083/connectors'
```

Update an existing instance.
```bash
curl -s -X PUT -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors/TestSinkConnector1/config
```