fix(sink): fix redis sink parse error (#15654)

Co-authored-by: xxhZs <[email protected]> Co-authored-by: Bugen Zhao <[email protected]>
risingwavelabs · Mar 14, 2024 · c002073 · c002073
1 parent 89c1152
commit c002073
Show file tree

Hide file tree

Showing 5 changed files with 9 additions and 9 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/src/connector/Cargo.toml b/src/connector/Cargo.toml
@@ -106,7 +106,7 @@ rdkafka = { workspace = true, features = [
     "gssapi",
     "zstd",
 ] }
-redis = { version = "0.24.0", features = [
+redis = { version = "0.25", features = [
     "aio",
     "tokio-comp",
     "async-std-comp",

diff --git a/src/connector/src/sink/redis.rs b/src/connector/src/sink/redis.rs
@@ -16,7 +16,7 @@ use std::collections::{HashMap, HashSet};
 
 use anyhow::anyhow;
 use async_trait::async_trait;
-use redis::aio::Connection;
+use redis::aio::MultiplexedConnection;
 use redis::{Client as RedisClient, Pipeline};
 use risingwave_common::array::StreamChunk;
 use risingwave_common::catalog::Schema;
@@ -170,14 +170,14 @@ pub struct RedisSinkWriter {
 
 struct RedisSinkPayloadWriter {
     // connection to redis, one per executor
-    conn: Option<Connection>,
+    conn: Option<MultiplexedConnection>,
     // 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_async_connection().await?);
+        let conn = Some(client.get_multiplexed_async_connection().await?);
         let pipe = redis::pipe();
 
         Ok(Self { conn, pipe })

diff --git a/src/risedevtool/Cargo.toml b/src/risedevtool/Cargo.toml
@@ -24,7 +24,7 @@ google-cloud-pubsub = "0.23"
 indicatif = "0.17"
 itertools = "0.12"
 rdkafka = { workspace = true }
-redis = "0.24"
+redis = "0.25"
 regex = "1"
 reqwest = "0.11"
 serde = { version = "1", features = ["derive"] }

diff --git a/src/workspace-hack/Cargo.toml b/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.24", features = ["async-std-comp", "tokio-comp"] }
+redis = { version = "0.25", features = ["async-std-comp", "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" }