From cee64a31438018b2005344fed3caf0cef596803f Mon Sep 17 00:00:00 2001 From: sumitaich1998 Date: Tue, 20 Sep 2022 11:09:28 +0530 Subject: [PATCH 1/8] docs: redis sink docs --- docs/README.md | 1 + docs/SUMMARY.md | 1 + docs/reference/configuration/redis.md | 90 +++++++++++++++++++++++++++ docs/roadmap.md | 2 +- docs/sinks/redis.md | 16 +++++ 5 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 docs/reference/configuration/redis.md create mode 100644 docs/sinks/redis.md diff --git a/docs/README.md b/docs/README.md index ea753403..31a55f8d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -10,6 +10,7 @@ GRPC) * Instrumentation support with statsd * Log Sink * Bigquery Sink +* Redis Sink Depot is a sink connector, which acts as a bridge between data processing systems and real sink. The APIs in this library can be used to push data to various sinks. Common sinks implementations will be added in this repo. diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index c79dc5af..c71246e5 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -5,6 +5,7 @@ ## Sink * [Bigquery](sinks/bigquery.md) +* [Redis](sinks/redis.md) ## Reference diff --git a/docs/reference/configuration/redis.md b/docs/reference/configuration/redis.md new file mode 100644 index 00000000..6d0e98f3 --- /dev/null +++ b/docs/reference/configuration/redis.md @@ -0,0 +1,90 @@ +# Redis + +A Redis sink Firehose (`SINK_TYPE`=`redis`) requires the following variables to be set along with Generic ones + +### `SINK_REDIS_URLS` + +REDIS instance hostname/IP address followed by its port. + +- Example value: `localhos:6379,localhost:6380` +- Type: `required` + +### `SINK_REDIS_DATA_TYPE` + +To select whether you want to push your data as a HashSet or as a List. + +- Example value: `Hashset` +- Type: `required` +- Default value: `Hashset` + +### `SINK_REDIS_KEY_TEMPLATE` + +The string that will act as the key for each Redis entry. This key can be configured as per the requirement, a constant or can extract value from each message and use that as the Redis key. + +- Example value: `Service\_%%s,1` + + This will take the value with index 1 from proto and create the Redis keys as per the template\ + +- Type: `required` + +### `SINK_REDIS_HASHSET_FIELD_TO_COLUMN_MAPPING` + +This is the field that decides what all data will be stored in the HashSet for each message. + +- Example value: `{"6":"customer_id", "2":"order_num"}` +- Type: `required (For Hashset)` + +### `SINK_REDIS_LIST_DATA_PROTO_INDEX` + +This field decides what all data will be stored in the List for each message. + +- Example value: `6` + + This will get the value of the field with index 6 in your proto and push that to the Redis list with the corresponding keyTemplate + +- Type: `required (For List)` + +### `SINK_REDIS_KEY_VALUE_DATA_FIELD_NAME` + +This field decides what data will be stored in the value part of key-value pair + +- Example value: `customer_id` + + This will get the value of the field with name `customer_id` in your proto and push that to the Redis as value with the corresponding keyTemplate + +- Type: `required (For KeyValue)` + +### `SINK_REDIS_LIST_DATA_FIELD_NAME` + +This field decides what all data will be stored in the List for each message. + +- Example value: `customer_id` + + This will get the value of the field with name `customer_id` in your proto and push that to the Redis list with the corresponding keyTemplate + +- Type: `required (For List)` + +### `SINK_REDIS_TTL_TYPE` + +- Example value: `DURATION` +- Type: `optional` +- Default value: `DISABLE` +- Choice of Redis TTL type.It can be:\ + - `DURATION`: After which the Key will be expired and removed from Redis \(UNIT- seconds\)\ + - `EXACT_TIME`: Precise UNIX timestamp after which the Key will be expired + +### `SINK_REDIS_TTL_VALUE` + +Redis TTL value in Unix Timestamp for `EXACT_TIME` TTL type, In Seconds for `DURATION` TTL type. + +- Example value: `100000` +- Type: `optional` +- Default value: `0` + +### `SINK_REDIS_DEPLOYMENT_TYPE` + +The Redis deployment you are using. At present, we support `Standalone` and `Cluster` types. + +- Example value: `Standalone` +- Type: `required` +- Default value: `Standalone` diff --git a/docs/roadmap.md b/docs/roadmap.md index 568d84b2..1eef242e 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -12,7 +12,7 @@ If you have feedback about the roadmap section itself, such as how the issues are presented, let us know through [discussions](https://github.com/odpf/depot/discussions). -## depot 0.1.4 +## Depot 0.1.4 ### Feature enhancements diff --git a/docs/sinks/redis.md b/docs/sinks/redis.md new file mode 100644 index 00000000..09400b09 --- /dev/null +++ b/docs/sinks/redis.md @@ -0,0 +1,16 @@ +# Redis Sink + +### Input Schema +Redis sink in Depot only supports Protobuf schema as of now. Support for JSON and Avro format is planned for a future release of Depot. + +### Data Types +Redis sink can be created in 2 different modes based on the value of [`SINK_REDIS_DATA_TYPE`](../sinks/redis-sink.md#sink_redis_data_type): HashSet or List + - `Hashset`: For each message, an entry of the format `key : field : value` is generated and pushed to Redis. field and value are generated on the basis of the config [`INPUT_SCHEMA_PROTO_TO_COLUMN_MAPPING`](../sinks/redis-sink.md#-input_schema_proto_to_column_mapping-2) + - `List`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto index provided in the config [`SINK_REDIS_LIST_DATA_PROTO_INDEX`](../sinks/redis-sink.md#sink_redis_list_data_proto_index) + +The `key` is picked up from a field in the message itself. + +Limitation: Depot Redis sink only supports HashSet and List entries as of now. + +### Deployment Types +Redis sink, as of now, supports two different Deployment Types `Standalone` and `Cluster`. This can be configured in the Depot environment variable `SINK_REDIS_DEPLOYMENT_TYPE`. From 1cb22f537ef926d7af042c7133c43296635b0c5f Mon Sep 17 00:00:00 2001 From: sumitaich1998 Date: Tue, 20 Sep 2022 11:53:32 +0530 Subject: [PATCH 2/8] docs: redis sink docs --- docs/reference/configuration/redis.md | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/docs/reference/configuration/redis.md b/docs/reference/configuration/redis.md index 6d0e98f3..e1955d66 100644 --- a/docs/reference/configuration/redis.md +++ b/docs/reference/configuration/redis.md @@ -1,6 +1,6 @@ # Redis -A Redis sink Firehose (`SINK_TYPE`=`redis`) requires the following variables to be set along with Generic ones +A Redis sink in Depot requires the following environment variables to be set along with Generic ones ### `SINK_REDIS_URLS` @@ -23,15 +23,14 @@ The string that will act as the key for each Redis entry. This key can be config - Example value: `Service\_%%s,1` - This will take the value with index 1 from proto and create the Redis keys as per the template\ + This will take the value with index 1 from proto and create the Redis keys as per the template. - Type: `required` ### `SINK_REDIS_HASHSET_FIELD_TO_COLUMN_MAPPING` This is the field that decides what all data will be stored in the HashSet for each message. - -- Example value: `{"6":"customer_id", "2":"order_num"}` +- Example value: `{"order_number":"ORDER_NUMBER","event_timestamp":"TIMESTAMP"}` - Type: `required (For Hashset)` ### `SINK_REDIS_LIST_DATA_PROTO_INDEX` @@ -69,8 +68,8 @@ This field decides what all data will be stored in the List for each message. - Example value: `DURATION` - Type: `optional` - Default value: `DISABLE` -- Choice of Redis TTL type.It can be:\ - - `DURATION`: After which the Key will be expired and removed from Redis \(UNIT- seconds\)\ +- Choice of Redis TTL type.It can be: + - `DURATION`: After which the Key will be expired and removed from Redis \(UNIT- seconds\) - `EXACT_TIME`: Precise UNIX timestamp after which the Key will be expired ### `SINK_REDIS_TTL_VALUE` From e0e6184467eed82a4ddb83ef50ff2d403d5c1d71 Mon Sep 17 00:00:00 2001 From: sumitaich1998 Date: Wed, 21 Sep 2022 11:06:33 +0530 Subject: [PATCH 3/8] docs: redis sink docs --- docs/sinks/redis.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/sinks/redis.md b/docs/sinks/redis.md index 09400b09..2b67d8c4 100644 --- a/docs/sinks/redis.md +++ b/docs/sinks/redis.md @@ -1,16 +1,16 @@ # Redis Sink ### Input Schema -Redis sink in Depot only supports Protobuf schema as of now. Support for JSON and Avro format is planned for a future release of Depot. +Redis sink in Depot only supports Protobuf and JSON schema as of now. Support for Avro format is planned for a future release of Depot. ### Data Types -Redis sink can be created in 2 different modes based on the value of [`SINK_REDIS_DATA_TYPE`](../sinks/redis-sink.md#sink_redis_data_type): HashSet or List - - `Hashset`: For each message, an entry of the format `key : field : value` is generated and pushed to Redis. field and value are generated on the basis of the config [`INPUT_SCHEMA_PROTO_TO_COLUMN_MAPPING`](../sinks/redis-sink.md#-input_schema_proto_to_column_mapping-2) - - `List`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto index provided in the config [`SINK_REDIS_LIST_DATA_PROTO_INDEX`](../sinks/redis-sink.md#sink_redis_list_data_proto_index) +Redis sink can be created in 2 different modes based on the value of [`SINK_REDIS_DATA_TYPE`](../reference/configuration/redis.md#sink_redis_data_type): HashSet or List + - `Hashset`: For each message, an entry of the format `key : field : value` is generated and pushed to Redis. Field and value are generated on the basis of the config [`INPUT_SCHEMA_PROTO_TO_COLUMN_MAPPING`](../reference/configuration/redis.md#sink_redis_hashset_field_to_column_mapping) + - `List`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto index provided in the config [`SINK_REDIS_LIST_DATA_PROTO_INDEX`](../reference/configuration/redis.md#sink_redis_list_data_proto_index) The `key` is picked up from a field in the message itself. -Limitation: Depot Redis sink only supports HashSet and List entries as of now. +Limitation: Depot Redis sink only supports Key-Value, HashSet and List entries as of now. ### Deployment Types Redis sink, as of now, supports two different Deployment Types `Standalone` and `Cluster`. This can be configured in the Depot environment variable `SINK_REDIS_DEPLOYMENT_TYPE`. From 486ed6696591c7ae03aae83afecfe1c211b27117 Mon Sep 17 00:00:00 2001 From: sumitaich1998 Date: Wed, 21 Sep 2022 11:09:46 +0530 Subject: [PATCH 4/8] docs: redis sink docs --- docs/sinks/redis.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/sinks/redis.md b/docs/sinks/redis.md index 2b67d8c4..c0040e20 100644 --- a/docs/sinks/redis.md +++ b/docs/sinks/redis.md @@ -5,7 +5,7 @@ Redis sink in Depot only supports Protobuf and JSON schema as of now. Support f ### Data Types Redis sink can be created in 2 different modes based on the value of [`SINK_REDIS_DATA_TYPE`](../reference/configuration/redis.md#sink_redis_data_type): HashSet or List - - `Hashset`: For each message, an entry of the format `key : field : value` is generated and pushed to Redis. Field and value are generated on the basis of the config [`INPUT_SCHEMA_PROTO_TO_COLUMN_MAPPING`](../reference/configuration/redis.md#sink_redis_hashset_field_to_column_mapping) + - `Hashset`: For each message, an entry of the format `key : field : value` is generated and pushed to Redis. Field and value are generated on the basis of the config [`SINK_REDIS_HASHSET_FIELD_TO_COLUMN_MAPPING`](../reference/configuration/redis.md#sink_redis_hashset_field_to_column_mapping) - `List`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto index provided in the config [`SINK_REDIS_LIST_DATA_PROTO_INDEX`](../reference/configuration/redis.md#sink_redis_list_data_proto_index) The `key` is picked up from a field in the message itself. From 1dce2b999f1b2a5905d73a0769ff7ca134d0847c Mon Sep 17 00:00:00 2001 From: sumitaich1998 Date: Wed, 21 Sep 2022 11:36:34 +0530 Subject: [PATCH 5/8] docs: add keyvalue mode in redis --- docs/reference/configuration/redis.md | 2 +- docs/sinks/redis.md | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/docs/reference/configuration/redis.md b/docs/reference/configuration/redis.md index e1955d66..fb046a5e 100644 --- a/docs/reference/configuration/redis.md +++ b/docs/reference/configuration/redis.md @@ -11,7 +11,7 @@ REDIS instance hostname/IP address followed by its port. ### `SINK_REDIS_DATA_TYPE` -To select whether you want to push your data as a HashSet or as a List. +To select whether you want to push your data as a `KEYVALUE`, `HASHSET` or as a `LIST`. - Example value: `Hashset` - Type: `required` diff --git a/docs/sinks/redis.md b/docs/sinks/redis.md index c0040e20..badea84d 100644 --- a/docs/sinks/redis.md +++ b/docs/sinks/redis.md @@ -4,9 +4,10 @@ Redis sink in Depot only supports Protobuf and JSON schema as of now. Support for Avro format is planned for a future release of Depot. ### Data Types -Redis sink can be created in 2 different modes based on the value of [`SINK_REDIS_DATA_TYPE`](../reference/configuration/redis.md#sink_redis_data_type): HashSet or List +Redis sink can be created in 3 different modes based on the value of [`SINK_REDIS_DATA_TYPE`](../reference/configuration/redis.md#sink_redis_data_type): HashSet, KeyValue or List - `Hashset`: For each message, an entry of the format `key : field : value` is generated and pushed to Redis. Field and value are generated on the basis of the config [`SINK_REDIS_HASHSET_FIELD_TO_COLUMN_MAPPING`](../reference/configuration/redis.md#sink_redis_hashset_field_to_column_mapping) - - `List`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto index provided in the config [`SINK_REDIS_LIST_DATA_PROTO_INDEX`](../reference/configuration/redis.md#sink_redis_list_data_proto_index) + - `List`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto index provided in the config [`SINK_REDIS_LIST_DATA_PROTO_INDEX`](../reference/configuration/redis.md#sink_redis_list_data_proto_index) or the Proto field name provided in the config [`SINK_REDIS_LIST_DATA_FIELD_NAME`](../reference/configuration/redis.md#sink_redis_list_data_field_name) + - `KeyValue`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto field name provided in the config [`SINK_REDIS_KEY_VALUE_DATA_FIELD_NAME`](../reference/configuration/redis.md#sink_redis_key_value_data_field_name) The `key` is picked up from a field in the message itself. From 7daa98006196bf79b21c279f89dbb76635f68cc9 Mon Sep 17 00:00:00 2001 From: sumitaich1998 Date: Wed, 21 Sep 2022 12:09:59 +0530 Subject: [PATCH 6/8] docs: redis sink docs --- docs/reference/configuration/redis.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/reference/configuration/redis.md b/docs/reference/configuration/redis.md index fb046a5e..a2ec5d68 100644 --- a/docs/reference/configuration/redis.md +++ b/docs/reference/configuration/redis.md @@ -4,14 +4,14 @@ A Redis sink in Depot requires the following environment variables to be set alo ### `SINK_REDIS_URLS` -REDIS instance hostname/IP address followed by its port. +REDIS server instance hostname/IP address followed by its port. -- Example value: `localhos:6379,localhost:6380` +- Example value: `localhost:6379,localhost:6380` - Type: `required` ### `SINK_REDIS_DATA_TYPE` -To select whether you want to push your data as a `KEYVALUE`, `HASHSET` or as a `LIST`. +To select whether you want to push your data as a `KEYVALUE`, `HASHSET` or as a `LIST` data type. - Example value: `Hashset` - Type: `required` From 5a5b41bac6b6b04aa58219158e79d0b49faec0bf Mon Sep 17 00:00:00 2001 From: sumitaich1998 Date: Wed, 21 Sep 2022 15:22:36 +0530 Subject: [PATCH 7/8] docs: redis sink docs --- docs/reference/configuration/redis.md | 10 ---------- docs/sinks/redis.md | 7 ++----- .../io/odpf/depot/config/RedisSinkConfig.java | 3 --- .../io/odpf/depot/redis/RedisSinkFactory.java | 17 ++++++++++++++--- 4 files changed, 16 insertions(+), 21 deletions(-) diff --git a/docs/reference/configuration/redis.md b/docs/reference/configuration/redis.md index a2ec5d68..22b296d2 100644 --- a/docs/reference/configuration/redis.md +++ b/docs/reference/configuration/redis.md @@ -33,16 +33,6 @@ This is the field that decides what all data will be stored in the HashSet for e - Example value: `{"order_number":"ORDER_NUMBER","event_timestamp":"TIMESTAMP"}` - Type: `required (For Hashset)` -### `SINK_REDIS_LIST_DATA_PROTO_INDEX` - -This field decides what all data will be stored in the List for each message. - -- Example value: `6` - - This will get the value of the field with index 6 in your proto and push that to the Redis list with the corresponding keyTemplate - -- Type: `required (For List)` - ### `SINK_REDIS_KEY_VALUE_DATA_FIELD_NAME` This field decides what data will be stored in the value part of key-value pair diff --git a/docs/sinks/redis.md b/docs/sinks/redis.md index badea84d..7b61e479 100644 --- a/docs/sinks/redis.md +++ b/docs/sinks/redis.md @@ -1,13 +1,10 @@ # Redis Sink -### Input Schema -Redis sink in Depot only supports Protobuf and JSON schema as of now. Support for Avro format is planned for a future release of Depot. - ### Data Types Redis sink can be created in 3 different modes based on the value of [`SINK_REDIS_DATA_TYPE`](../reference/configuration/redis.md#sink_redis_data_type): HashSet, KeyValue or List - `Hashset`: For each message, an entry of the format `key : field : value` is generated and pushed to Redis. Field and value are generated on the basis of the config [`SINK_REDIS_HASHSET_FIELD_TO_COLUMN_MAPPING`](../reference/configuration/redis.md#sink_redis_hashset_field_to_column_mapping) - - `List`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto index provided in the config [`SINK_REDIS_LIST_DATA_PROTO_INDEX`](../reference/configuration/redis.md#sink_redis_list_data_proto_index) or the Proto field name provided in the config [`SINK_REDIS_LIST_DATA_FIELD_NAME`](../reference/configuration/redis.md#sink_redis_list_data_field_name) - - `KeyValue`: For each message entry of the format `key : value` is generated and pushed to Redis. Value is fetched for the proto field name provided in the config [`SINK_REDIS_KEY_VALUE_DATA_FIELD_NAME`](../reference/configuration/redis.md#sink_redis_key_value_data_field_name) + - `List`: For each message entry of the format `key : value` is generated and pushed to Redis. The value is fetched for the Proto field name provided in the config [`SINK_REDIS_LIST_DATA_FIELD_NAME`](../reference/configuration/redis.md#sink_redis_list_data_field_name) + - `KeyValue`: For each message entry of the format `key : value` is generated and pushed to Redis. The value is fetched for the proto field name provided in the config [`SINK_REDIS_KEY_VALUE_DATA_FIELD_NAME`](../reference/configuration/redis.md#sink_redis_key_value_data_field_name) The `key` is picked up from a field in the message itself. diff --git a/src/main/java/io/odpf/depot/config/RedisSinkConfig.java b/src/main/java/io/odpf/depot/config/RedisSinkConfig.java index cdd4731c..dfe0eb57 100644 --- a/src/main/java/io/odpf/depot/config/RedisSinkConfig.java +++ b/src/main/java/io/odpf/depot/config/RedisSinkConfig.java @@ -39,9 +39,6 @@ public interface RedisSinkConfig extends OdpfSinkConfig { @ConverterClass(RedisSinkDeploymentTypeConverter.class) RedisSinkDeploymentType getSinkRedisDeploymentType(); - @Key("SINK_REDIS_LIST_DATA_PROTO_INDEX") - String getSinkRedisListDataProtoIndex(); - @Key("SINK_REDIS_KEY_VALUE_DATA_FIELD_NAME") String getSinkRedisKeyValueDataFieldName(); diff --git a/src/main/java/io/odpf/depot/redis/RedisSinkFactory.java b/src/main/java/io/odpf/depot/redis/RedisSinkFactory.java index d8132ebb..f718a98f 100644 --- a/src/main/java/io/odpf/depot/redis/RedisSinkFactory.java +++ b/src/main/java/io/odpf/depot/redis/RedisSinkFactory.java @@ -36,14 +36,25 @@ public RedisSinkFactory(RedisSinkConfig sinkConfig) { public void init() throws IOException { Instrumentation instrumentation = new Instrumentation(statsDReporter, RedisSinkFactory.class); - String redisConfig = String.format("\n\tredis.urls = %s\n\tredis.key.template = %s\n\tredis.sink.type = %s" - + "\n\tredis.list.data.proto.index = %s\n\tredis.ttl.type = %s\n\tredis.ttl.value = %d", + String redisConfig = String.format("\n\tredis.urls = %s\n\tredis.key.template = %s\n\tredis.sink.data.type = %s" + + "\n\tredis.deployment.type = %s\n\tredis.ttl.type = %s\n\tredis.ttl.value = %d\n\t", sinkConfig.getSinkRedisUrls(), sinkConfig.getSinkRedisKeyTemplate(), sinkConfig.getSinkRedisDataType().toString(), - sinkConfig.getSinkRedisListDataProtoIndex(), + sinkConfig.getSinkRedisDeploymentType().toString(), sinkConfig.getSinkRedisTtlType().toString(), sinkConfig.getSinkRedisTtlValue()); + switch (sinkConfig.getSinkRedisDataType()) { + case LIST: + redisConfig += "redis.list.data.field.name=" + sinkConfig.getSinkRedisListDataFieldName(); + break; + case KEYVALUE: + redisConfig += "redis.keyvalue.data.field.name=" + sinkConfig.getSinkRedisKeyValueDataFieldName(); + break; + case HASHSET: + redisConfig += "redis.hashset.field.to.column.mapping=" + sinkConfig.getSinkRedisHashsetFieldToColumnMapping().toString(); + break; + } instrumentation.logInfo(redisConfig); instrumentation.logInfo("Redis server type = {}", sinkConfig.getSinkRedisDeploymentType()); OdpfMessageParser messageParser = OdpfMessageParserFactory.getParser(sinkConfig, statsDReporter); From 0af98df9db776ab97683a7fb4808ec388f9491bd Mon Sep 17 00:00:00 2001 From: sumitaich1998 Date: Wed, 21 Sep 2022 15:27:33 +0530 Subject: [PATCH 8/8] docs: redis sink docs --- src/main/java/io/odpf/depot/redis/RedisSinkFactory.java | 1 + 1 file changed, 1 insertion(+) diff --git a/src/main/java/io/odpf/depot/redis/RedisSinkFactory.java b/src/main/java/io/odpf/depot/redis/RedisSinkFactory.java index f718a98f..42ffc6e3 100644 --- a/src/main/java/io/odpf/depot/redis/RedisSinkFactory.java +++ b/src/main/java/io/odpf/depot/redis/RedisSinkFactory.java @@ -54,6 +54,7 @@ public void init() throws IOException { case HASHSET: redisConfig += "redis.hashset.field.to.column.mapping=" + sinkConfig.getSinkRedisHashsetFieldToColumnMapping().toString(); break; + default: } instrumentation.logInfo(redisConfig); instrumentation.logInfo("Redis server type = {}", sinkConfig.getSinkRedisDeploymentType());