fix(sink): Fix redis commit bug (#13538)

risingwavelabs · Nov 23, 2023 · 0ecc838 · 0ecc838
1 parent edd0805
commit 0ecc838
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/src/connector/src/sink/redis.rs b/src/connector/src/sink/redis.rs
@@ -190,6 +190,12 @@ impl RedisSinkPayloadWriter {
     }
 
     pub async fn commit(&mut self) -> Result<()> {
+        #[cfg(test)]
+        {
+            if self.conn.is_none() {
+                return Ok(());
+            }
+        }
         self.pipe.query_async(self.conn.as_mut().unwrap()).await?;
         self.pipe.clear();
         Ok(())
@@ -271,7 +277,8 @@ impl AsyncTruncateSinkWriter for RedisSinkWriter {
         _add_future: DeliveryFutureManagerAddFuture<'a, Self::DeliveryFuture>,
     ) -> Result<()> {
         dispatch_sink_formatter_str_key_impl!(&self.formatter, formatter, {
-            self.payload_writer.write_chunk(chunk, formatter).await
+            self.payload_writer.write_chunk(chunk, formatter).await?;
+            self.payload_writer.commit().await
         })
     }
 }