diff --git a/src/connector/src/sink/encoder/base64_adapter.rs b/src/connector/src/sink/encoder/base64_adapter.rs
index 9c47c0a8f91da..77ab65c7e2e31 100644
--- a/src/connector/src/sink/encoder/base64_adapter.rs
+++ b/src/connector/src/sink/encoder/base64_adapter.rs
@@ -38,8 +38,8 @@ impl<E: RowEncoder> RowEncoder for Base64Adapter<E> {
 }
 
 impl SerToString for String {
-    fn ser_to_string(&self) -> Result<String> {
-        Ok(self.clone())
+    fn ser_to_string(self) -> Result<String> {
+        Ok(self)
     }
 }
 
diff --git a/src/connector/src/sink/encoder/empty.rs b/src/connector/src/sink/encoder/empty.rs
index 53c66cf8af354..aa65a5a7a336f 100644
--- a/src/connector/src/sink/encoder/empty.rs
+++ b/src/connector/src/sink/encoder/empty.rs
@@ -31,7 +31,7 @@ impl RowEncoder for EmptyEncoder {
 }
 
 impl SerToString for () {
-    fn ser_to_string(&self) -> Result<String> {
+    fn ser_to_string(self) -> Result<String> {
         Ok("".to_string())
     }
 }
diff --git a/src/connector/src/sink/encoder/json.rs b/src/connector/src/sink/encoder/json.rs
index 920c87c34f016..443029b375210 100644
--- a/src/connector/src/sink/encoder/json.rs
+++ b/src/connector/src/sink/encoder/json.rs
@@ -60,13 +60,13 @@ impl RowEncoder for JsonEncoder {
 }
 
 impl SerToString for Map<String, Value> {
-    fn ser_to_string(&self) -> Result<String> {
-        serde_json::to_string(self).map_err(|e| SinkError::JsonParse(e.to_string()))
+    fn ser_to_string(self) -> Result<String> {
+        serde_json::to_string(&self).map_err(|e| SinkError::JsonParse(e.to_string()))
     }
 }
 
 impl SerToString for Value {
-    fn ser_to_string(&self) -> Result<String> {
+    fn ser_to_string(self) -> Result<String> {
         Ok(self.to_string())
     }
 }
diff --git a/src/connector/src/sink/encoder/mod.rs b/src/connector/src/sink/encoder/mod.rs
index dff65ad36ebc0..94537a49919bd 100644
--- a/src/connector/src/sink/encoder/mod.rs
+++ b/src/connector/src/sink/encoder/mod.rs
@@ -45,15 +45,15 @@ pub trait RowEncoder {
 }
 
 pub trait SerToBytes {
-    fn ser_to_bytes(&self) -> Result<Vec<u8>>;
+    fn ser_to_bytes(self) -> Result<Vec<u8>>;
 }
 
 pub trait SerToString {
-    fn ser_to_string(&self) -> Result<String>;
+    fn ser_to_string(self) -> Result<String>;
 }
 
 impl<T: SerToString> SerToBytes for T {
-    fn ser_to_bytes(&self) -> Result<Vec<u8>> {
+    fn ser_to_bytes(self) -> Result<Vec<u8>> {
         self.ser_to_string().map(|s| s.into_bytes())
     }
 }
diff --git a/src/connector/src/sink/encoder/proto.rs b/src/connector/src/sink/encoder/proto.rs
index 79db6597cc965..617af2cfec9d2 100644
--- a/src/connector/src/sink/encoder/proto.rs
+++ b/src/connector/src/sink/encoder/proto.rs
@@ -36,7 +36,7 @@ impl RowEncoder for ProtoEncoder {
 }
 
 impl SerToBytes for DynamicMessage {
-    fn ser_to_bytes(&self) -> Result<Vec<u8>> {
+    fn ser_to_bytes(self) -> Result<Vec<u8>> {
         Ok(self.encode_to_vec())
     }
 }
diff --git a/src/connector/src/sink/mod.rs b/src/connector/src/sink/mod.rs
index ce216ff4bd53c..1d838d80db2c2 100644
--- a/src/connector/src/sink/mod.rs
+++ b/src/connector/src/sink/mod.rs
@@ -241,15 +241,15 @@ pub trait MessageSink {
     }
 }
 pub trait SerTo<T> {
-    fn ser_to(&self) -> Result<T>;
+    fn ser_to(self) -> Result<T>;
 }
 impl<T: SerToBytes> SerTo<Vec<u8>> for T {
-    fn ser_to(&self) -> Result<Vec<u8>> {
+    fn ser_to(self) -> Result<Vec<u8>> {
         self.ser_to_bytes()
     }
 }
 impl<T: SerToString> SerTo<String> for T {
-    fn ser_to(&self) -> Result<String> {
+    fn ser_to(self) -> Result<String> {
         self.ser_to_string()
     }
 }