Introduction

The Redis plugin is a collection of connectors that are used to interact with a Redis cluster.

Sink Connectors

Redis Sink Connector

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 ByteArrayConverter or the 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.client.mode

The client mode to use when interacting with the Redis cluster.

Importance: Medium

Type: String

Default Value: Standalone

Validator: Matches: Standalone, Cluster

redis.database

Redis database to connect to.

Importance: Medium

Type: Int

Default Value: 1

redis.operation.timeout.ms

The amount of time in milliseconds before an operation is marked as timed out.

Importance: Medium

Type: Long

Default Value: 10000

Validator: [100,...]

redis.password

Password used to connect to Redis.

Importance: Medium

Type: Password

Default Value: [hidden]

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

Default Value: [hidden]

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.charset

The character set to use for String key and values.

Importance: Low

Type: String

Default Value: UTF-8

Validator: Valid values: 'Big5', 'Big5-HKSCS', 'CESU-8', 'EUC-JP', 'EUC-KR', 'GB18030', 'GB2312', 'GBK', 'IBM-Thai', 'IBM00858', 'IBM01140', 'IBM01141', 'IBM01142', 'IBM01143', 'IBM01144', 'IBM01145', 'IBM01146', 'IBM01147', 'IBM01148', 'IBM01149', 'IBM037', 'IBM1026', 'IBM1047', 'IBM273', 'IBM277', 'IBM278', 'IBM280', 'IBM284', 'IBM285', 'IBM290', 'IBM297', 'IBM420', 'IBM424', 'IBM437', 'IBM500', 'IBM775', 'IBM850', 'IBM852', 'IBM855', 'IBM857', 'IBM860', 'IBM861', 'IBM862', 'IBM863', 'IBM864', 'IBM865', 'IBM866', 'IBM868', 'IBM869', 'IBM870', 'IBM871', 'IBM918', 'ISO-2022-CN', 'ISO-2022-JP', 'ISO-2022-JP-2', 'ISO-2022-KR', 'ISO-8859-1', 'ISO-8859-13', 'ISO-8859-15', 'ISO-8859-2', 'ISO-8859-3', 'ISO-8859-4', 'ISO-8859-5', 'ISO-8859-6', 'ISO-8859-7', 'ISO-8859-8', 'ISO-8859-9', 'JIS_X0201', 'JIS_X0212-1990', 'KOI8-R', 'KOI8-U', 'Shift_JIS', 'TIS-620', 'US-ASCII', 'UTF-16', 'UTF-16BE', 'UTF-16LE', 'UTF-32', 'UTF-32BE', 'UTF-32LE', 'UTF-8', 'X-UTF-32BE-BOM', 'X-UTF-32LE-BOM', 'windows-1250', 'windows-1251', 'windows-1252', 'windows-1253', 'windows-1254', 'windows-1255', 'windows-1256', 'windows-1257', 'windows-1258', 'windows-31j', 'x-Big5-HKSCS-2001', 'x-Big5-Solaris', 'x-COMPOUND_TEXT', 'x-EUC-TW', 'x-IBM1006', 'x-IBM1025', 'x-IBM1046', 'x-IBM1097', 'x-IBM1098', 'x-IBM1112', 'x-IBM1122', 'x-IBM1123', 'x-IBM1124', 'x-IBM1364', 'x-IBM1381', 'x-IBM1383', 'x-IBM300', 'x-IBM33722', 'x-IBM737', 'x-IBM833', 'x-IBM834', 'x-IBM856', 'x-IBM874', 'x-IBM875', 'x-IBM921', 'x-IBM922', 'x-IBM930', 'x-IBM933', 'x-IBM935', 'x-IBM937', 'x-IBM939', 'x-IBM942', 'x-IBM942C', 'x-IBM943', 'x-IBM943C', 'x-IBM948', 'x-IBM949', 'x-IBM949C', 'x-IBM950', 'x-IBM964', 'x-IBM970', 'x-ISCII91', 'x-ISO-2022-CN-CNS', 'x-ISO-2022-CN-GB', 'x-JIS0208', 'x-JISAutoDetect', 'x-Johab', 'x-MS932_0213', 'x-MS950-HKSCS', 'x-MS950-HKSCS-XP', 'x-MacArabic', 'x-MacCentralEurope', 'x-MacCroatian', 'x-MacCyrillic', 'x-MacDingbat', 'x-MacGreek', 'x-MacHebrew', 'x-MacIceland', 'x-MacRoman', 'x-MacRomania', 'x-MacSymbol', 'x-MacThai', 'x-MacTurkish', 'x-MacUkraine', 'x-PCK', 'x-SJIS_0213', 'x-UTF-16LE-BOM', 'x-euc-jp-linux', 'x-eucJP-Open', 'x-iso-8859-11', 'x-mswin-936', 'x-windows-50220', 'x-windows-50221', 'x-windows-874', 'x-windows-949', 'x-windows-950', 'x-windows-iso2022jp'

redis.request.queue.size

The maximum number of queued requests to Redis.

Importance: Low

Type: Int

Default Value: 2147483647

redis.socket.connect.timeout.ms

The amount of time in milliseconds to wait before timing out a socket when connecting.

Importance: Low

Type: Int

Default Value: 10000

redis.socket.keep.alive.enabled

Flag to enable a keepalive to Redis.

Importance: Low

Type: Boolean

Default Value: false

redis.socket.tcp.no.delay.enabled

Flag to enable TCP no delay should be used.

Importance: Low

Type: Boolean

Default Value: true

redis.ssl.provider

The SSL provider to use.

Importance: Low

Type: String

Default Value: JDK

Validator: Matches: OPENSSL, JDK

Examples

Standalone Example

This configuration is used typically along with standalone mode.

name=RedisSinkConnector1
connector.class=com.github.jcustenborder.kafka.connect.redis.RedisSinkConnector
tasks.max=1
topics=< Required Configuration >

Distributed Example

This configuration is used typically along with distributed mode. 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).

{
  "config" : {
    "name" : "RedisSinkConnector1",
    "connector.class" : "com.github.jcustenborder.kafka.connect.redis.RedisSinkConnector",
    "tasks.max" : "1",
    "topics" : "< Required Configuration >"
  }
}

Use 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.

curl -s -X POST -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors

Update an existing instance.

curl -s -X PUT -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors/TestSinkConnector1/config