feat(sink): support redis cluster url

ci/redis-conf/redis-7000.conf

@@ -0,0 +1,9 @@
+port 7000
+
+#enable cluster mode
+cluster-enabled yes
+
+#ms
+cluster-node-timeout 15000
+
+cluster-config-file "nodes-7000.conf"

ci/redis-conf/redis-7001.conf

@@ -0,0 +1,9 @@
+port 7001
+
+#enable cluster mode
+cluster-enabled yes
+
+#ms
+cluster-node-timeout 15000
+
+cluster-config-file "nodes-7001.conf"

ci/redis-conf/redis-7002.conf

@@ -0,0 +1,9 @@
+port 7002
+
+#enable cluster mode
+cluster-enabled yes
+
+#ms
+cluster-node-timeout 15000
+
+cluster-config-file "nodes-7002.conf"

ci/scripts/e2e-redis-sink-test.sh

@@ -44,5 +44,24 @@ else
   exit 1
 fi
 
+echo "--- testing cluster sinks"
+redis-server ./ci/redis-conf/redis-7000.conf --daemonize yes
+redis-server ./ci/redis-conf/redis-7001.conf --daemonize yes
+redis-server ./ci/redis-conf/redis-7002.conf --daemonize yes
+
+echo "yes" | redis-cli --cluster create 127.0.0.1:7000 127.0.0.1:7001 127.0.0.1:7002
+
+sqllogictest -p 4566 -d dev './e2e_test/sink/redis_cluster_sink.slt'
+
+redis-cli -c --cluster call 127.0.0.1:7000 keys \* >> ./query_result_1.txt
+
+if cat ./query_result_1.txt | tr '\n' '\0' | xargs -0 -n1 bash -c '[[ "$0" == "{\"v1\":1}" || "$0" == "{\"v2\":2}" || "$0" == "{\"v3\":3}" ]]'; then
+    echo "Redis sink check passed"
+else
+    cat ./query_result_1.txt
+  echo "The output is not as expected."
+  exit 1
+fi
+
 echo "--- Kill cluster"
 cargo make ci-kill

e2e_test/sink/redis_cluster_sink.slt

@@ -0,0 +1,35 @@
+statement ok
+CREATE TABLE t6 (v1 int primary key, v2 int);
+
+statement ok
+CREATE MATERIALIZED VIEW mv6 AS SELECT * FROM t6;
+
+statement ok
+CREATE SINK s61
+FROM
+    mv6 WITH (
+    primary_key = 'v1',
+    connector = 'redis',
+    redis.url= '["redis://redis-server:7000/","redis://redis-server:7001/","redis://redis-server:7002/"]',
+)FORMAT PLAIN ENCODE JSON(force_append_only='true');
+
+statement ok
+INSERT INTO t6 VALUES (1, 1);
+
+statement ok
+INSERT INTO t6 VALUES (2, 2);
+
+statement ok
+INSERT INTO t6 VALUES (3, 3);
+
+statement ok
+FLUSH;
+
+statement ok
+DROP SINK s61;
+
+statement ok
+DROP MATERIALIZED VIEW mv6;
+
+statement ok
+DROP TABLE t6;

e2e_test/sink/redis_sink.slt

@@ -10,7 +10,7 @@ FROM
     mv6 WITH (
     primary_key = 'v1',
     connector = 'redis',
-    redis.url= 'redis://redis-server:6379/',
+    redis.url= '["redis://redis-server:6379/"]',
 )FORMAT PLAIN ENCODE JSON(force_append_only='true');
 
 statement ok
@@ -19,7 +19,7 @@ FROM
     mv6 WITH (
     primary_key = 'v1',
     connector = 'redis',
-    redis.url= 'redis://redis-server:6379/',
+    redis.url= '["redis://redis-server:6379/"]',
 )FORMAT PLAIN ENCODE TEMPLATE(force_append_only='true', key_format = 'V1:{v1}', value_format = 'V2:{v2},V3:{v3}');
 
 statement ok

integration_tests/redis-sink/create_sink.sql

@@ -3,31 +3,31 @@ FROM
     bhv_mv WITH (
     primary_key = 'user_id',
     connector = 'redis',
-    redis.url= 'redis://redis:6379/',
+    redis.url= '["redis://127.0.0.1:6379/"]',
 )FORMAT PLAIN ENCODE JSON(force_append_only='true');
 
 CREATE SINK bhv_redis_sink_2
 FROM
     bhv_mv WITH (
     primary_key = 'user_id',
     connector = 'redis',
-    redis.url= 'redis://redis:6379/',
+    redis.url= '["redis://redis:6379/"]',
 )FORMAT PLAIN ENCODE TEMPLATE(force_append_only='true', key_format = 'UserID:{user_id}', value_format = 'TargetID:{target_id},EventTimestamp{event_timestamp}');
 
 CREATE SINK redis_types_json_sink
 FROM
     redis_types WITH (
     primary_key = 'types_id',
     connector = 'redis',
-    redis.url= 'redis://redis:6379/',
+    redis.url= '["redis://redis:6379/"]',
 )FORMAT PLAIN ENCODE JSON(force_append_only='true');
 
 CREATE SINK redis_types_template_sink
 FROM
     redis_types WITH (
     primary_key = 'types_id',
     connector = 'redis',
-    redis.url= 'redis://redis:6379/',
+    redis.url= '["redis://redis:6379/"]',
 )FORMAT PLAIN ENCODE TEMPLATE(force_append_only='true',
 key_format = 'TYPESID:{types_id}',
 value_format = '

src/connector/Cargo.toml

@@ -106,7 +106,7 @@ rdkafka = { workspace = true, features = [
     "gssapi",
     "zstd",
 ] }
-redis = { version = "0.25", features = ["aio", "tokio-comp", "async-std-comp"] }
+redis = { version = "0.25", features = ["aio", "tokio-comp", "async-std-comp","cluster-async"] }
 regex = "1.4"
 reqwest = { version = "0.11", features = ["json"] }
 risingwave_common = { workspace = true }

src/connector/src/sink/redis.rs

@@ -16,12 +16,16 @@ use std::collections::{HashMap, HashSet};
 
 use anyhow::anyhow;
 use async_trait::async_trait;
-use redis::aio::MultiplexedConnection;
+use redis::aio::{ConnectionLike, MultiplexedConnection};
+use redis::cluster::ClusterClient;
+use redis::cluster_async::ClusterConnection;
 use redis::{Client as RedisClient, Pipeline};
 use risingwave_common::array::StreamChunk;
 use risingwave_common::catalog::Schema;
 use serde_derive::Deserialize;
+use serde_json::Value;
 use serde_with::serde_as;
+use simd_json::prelude::ArrayTrait;
 use with_options::WithOptions;
 
 use super::catalog::SinkFormatDesc;
@@ -44,13 +48,57 @@ pub const VALUE_FORMAT: &str = "value_format";
 #[derive(Deserialize, Debug, Clone, WithOptions)]
 pub struct RedisCommon {
     #[serde(rename = "redis.url")]
-    pub url: String,
+    pub url: Vec<String>,
+}
+pub enum RedisConn {
+    Cluster(ClusterConnection),
+    Single(MultiplexedConnection),
+}
+
+impl ConnectionLike for RedisConn {
+    fn req_packed_command<'a>(
+        &'a mut self,
+        cmd: &'a redis::Cmd,
+    ) -> redis::RedisFuture<'a, redis::Value> {
+        match self {
+            RedisConn::Cluster(conn) => conn.req_packed_command(cmd),
+            RedisConn::Single(conn) => conn.req_packed_command(cmd),
+        }
+    }
+
+    fn req_packed_commands<'a>(
+        &'a mut self,
+        cmd: &'a redis::Pipeline,
+        offset: usize,
+        count: usize,
+    ) -> redis::RedisFuture<'a, Vec<redis::Value>> {
+        match self {
+            RedisConn::Cluster(conn) => conn.req_packed_commands(cmd, offset, count),
+            RedisConn::Single(conn) => conn.req_packed_commands(cmd, offset, count),
+        }
+    }
+
+    fn get_db(&self) -> i64 {
+        match self {
+            RedisConn::Cluster(conn) => conn.get_db(),
+            RedisConn::Single(conn) => conn.get_db(),
+        }
+    }
 }
 
 impl RedisCommon {
-    pub(crate) fn build_client(&self) -> ConnectorResult<RedisClient> {
-        let client = RedisClient::open(self.url.clone())?;
-        Ok(client)
+    pub async fn build_conn(&self) -> ConnectorResult<RedisConn> {
+        if self.url.is_empty() {
+            Err(SinkError::Redis("Missing redis.url".to_string()).into())
+        } else if self.url.len() == 1 {
+            let client = RedisClient::open(self.url.get(0).unwrap().clone())?;
+            Ok(RedisConn::Single(
+                client.get_multiplexed_async_connection().await?,
+            ))
+        } else {
+            let client = ClusterClient::new(self.url.clone())?;
+            Ok(RedisConn::Cluster(client.get_async_connection().await?))
+        }
     }
 }
 #[serde_as]
@@ -62,9 +110,13 @@ pub struct RedisConfig {
 
 impl RedisConfig {
     pub fn from_hashmap(properties: HashMap<String, String>) -> Result<Self> {
-        let config =
-            serde_json::from_value::<RedisConfig>(serde_json::to_value(properties).unwrap())
-                .map_err(|e| SinkError::Config(anyhow!(e)))?;
+        let mut json_map = serde_json::Map::new();
+        for (key, value) in properties {
+            let json_value = serde_json::from_str(&value).unwrap_or(Value::Null);
+            json_map.insert(key, json_value);
+        }
+        let config = serde_json::from_value::<RedisConfig>(Value::Object(json_map))
+            .map_err(|e| SinkError::Config(anyhow!(e)))?;
         Ok(config)
     }
 }
@@ -123,8 +175,7 @@ impl Sink for RedisSink {
     }
 
     async fn validate(&self) -> Result<()> {
-        let client = self.config.common.build_client()?;
-        client.get_connection()?;
+        let _conn = self.config.common.build_conn().await?;
         let all_set: HashSet<String> = self
             .schema
             .fields()
@@ -170,14 +221,14 @@ pub struct RedisSinkWriter {
 
 struct RedisSinkPayloadWriter {
     // connection to redis, one per executor
-    conn: Option<MultiplexedConnection>,
+    conn: Option<RedisConn>,
     // the command pipeline for write-commit
     pipe: Pipeline,
 }
 impl RedisSinkPayloadWriter {
     pub async fn new(config: RedisConfig) -> Result<Self> {
-        let client = config.common.build_client()?;
-        let conn = Some(client.get_multiplexed_async_connection().await?);
+        let conn = config.common.build_conn().await?;
+        let conn = Some(conn);
         let pipe = redis::pipe();
 
         Ok(Self { conn, pipe })

src/connector/with_options_sink.yaml

@@ -516,7 +516,7 @@ PulsarConfig:
 RedisConfig:
   fields:
   - name: redis.url
-    field_type: String
+    field_type: Vec<String>
     required: true
 StarrocksConfig:
   fields:

src/workspace-hack/Cargo.toml

@@ -108,7 +108,7 @@ quote = { version = "1" }
 rand = { version = "0.8", features = ["small_rng"] }
 rand_chacha = { version = "0.3" }
 rand_core = { version = "0.6", default-features = false, features = ["std"] }
-redis = { version = "0.25", features = ["async-std-comp", "tokio-comp"] }
+redis = { version = "0.25", features = ["async-std-comp", "cluster-async", "tokio-comp"] }
 regex = { version = "1" }
 regex-automata = { version = "0.4", default-features = false, features = ["dfa", "hybrid", "meta", "nfa", "perf", "unicode"] }
 regex-syntax = { version = "0.8" }