diff --git a/core/src/services/sftp/backend.rs b/core/src/services/sftp/backend.rs
index 839c1ffd8675..5be1f3bbe781 100644
--- a/core/src/services/sftp/backend.rs
+++ b/core/src/services/sftp/backend.rs
@@ -34,6 +34,8 @@ use serde::Deserialize;
 
 use super::error::is_not_found;
 use super::error::is_sftp_protocol_error;
+use super::error::parse_sftp_error;
+use super::error::parse_ssh_error;
 use super::lister::SftpLister;
 use super::writer::SftpWriter;
 use crate::raw::*;
@@ -291,7 +293,7 @@ impl Accessor for SftpBackend {
             if let Err(e) = res {
                 // ignore error if dir already exists
                 if !is_sftp_protocol_error(&e) {
-                    return Err(e.into());
+                    return Err(parse_sftp_error(e));
                 }
             }
             fs.set_cwd(&current);
@@ -305,7 +307,7 @@ impl Accessor for SftpBackend {
         let mut fs = client.fs();
         fs.set_cwd(&self.root);
 
-        let meta: Metadata = fs.metadata(path).await?.into();
+        let meta: Metadata = fs.metadata(path).await.map_err(parse_sftp_error)?.into();
 
         Ok(RpStat::new(meta))
     }
@@ -315,9 +317,12 @@ impl Accessor for SftpBackend {
 
         let mut fs = client.fs();
         fs.set_cwd(&self.root);
-        let path = fs.canonicalize(path).await?;
+        let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
 
-        let f = client.open(path.as_path()).await?;
+        let f = client
+            .open(path.as_path())
+            .await
+            .map_err(parse_sftp_error)?;
 
         // Sorry for the ugly code...
         //
@@ -339,7 +344,7 @@ impl Accessor for SftpBackend {
 
         let mut fs = client.fs();
         fs.set_cwd(&self.root);
-        let path = fs.canonicalize(path).await?;
+        let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
 
         let mut option = client.options();
         option.create(true);
@@ -349,7 +354,7 @@ impl Accessor for SftpBackend {
             option.write(true);
         }
 
-        let file = option.open(path).await?;
+        let file = option.open(path).await.map_err(parse_sftp_error)?;
 
         Ok((RpWrite::new(), SftpWriter::new(file)))
     }
@@ -368,7 +373,7 @@ impl Accessor for SftpBackend {
                     if is_not_found(&e) {
                         return Ok(RpDelete::default());
                     } else {
-                        return Err(e.into());
+                        return Err(parse_sftp_error(e));
                     }
                 }
             }
@@ -376,7 +381,7 @@ impl Accessor for SftpBackend {
             .boxed();
 
             while let Some(file) = dir.next().await {
-                let file = file?;
+                let file = file.map_err(parse_sftp_error)?;
                 let file_name = file.filename().to_str();
                 if file_name == Some(".") || file_name == Some("..") {
                     continue;
@@ -394,14 +399,14 @@ impl Accessor for SftpBackend {
 
             match fs.remove_dir(path).await {
                 Err(e) if !is_not_found(&e) => {
-                    return Err(e.into());
+                    return Err(parse_sftp_error(e));
                 }
                 _ => {}
             }
         } else {
             match fs.remove_file(path).await {
                 Err(e) if !is_not_found(&e) => {
-                    return Err(e.into());
+                    return Err(parse_sftp_error(e));
                 }
                 _ => {}
             }
@@ -423,7 +428,7 @@ impl Accessor for SftpBackend {
                 if is_not_found(&e) {
                     return Ok((RpList::default(), None));
                 } else {
-                    return Err(e.into());
+                    return Err(parse_sftp_error(e));
                 }
             }
         }
@@ -445,12 +450,15 @@ impl Accessor for SftpBackend {
             self.create_dir(dir, OpCreateDir::default()).await?;
         }
 
-        let src = fs.canonicalize(from).await?;
-        let dst = fs.canonicalize(to).await?;
-        let mut src_file = client.open(&src).await?;
-        let mut dst_file = client.create(dst).await?;
+        let src = fs.canonicalize(from).await.map_err(parse_sftp_error)?;
+        let dst = fs.canonicalize(to).await.map_err(parse_sftp_error)?;
+        let mut src_file = client.open(&src).await.map_err(parse_sftp_error)?;
+        let mut dst_file = client.create(dst).await.map_err(parse_sftp_error)?;
 
-        src_file.copy_all_to(&mut dst_file).await?;
+        src_file
+            .copy_all_to(&mut dst_file)
+            .await
+            .map_err(parse_sftp_error)?;
 
         Ok(RpCopy::default())
     }
@@ -464,7 +472,7 @@ impl Accessor for SftpBackend {
         if let Some((dir, _)) = to.rsplit_once('/') {
             self.create_dir(dir, OpCreateDir::default()).await?;
         }
-        fs.rename(from, to).await?;
+        fs.rename(from, to).await.map_err(parse_sftp_error)?;
 
         Ok(RpRename::default())
     }
@@ -517,9 +525,11 @@ async fn connect_sftp(
 
     session.known_hosts_check(known_hosts_strategy);
 
-    let session = session.connect(&endpoint).await?;
+    let session = session.connect(&endpoint).await.map_err(parse_ssh_error)?;
 
-    let sftp = Sftp::from_session(session, SftpOptions::default()).await?;
+    let sftp = Sftp::from_session(session, SftpOptions::default())
+        .await
+        .map_err(parse_sftp_error)?;
 
     if !root.is_empty() {
         let mut fs = sftp.fs();
@@ -533,7 +543,7 @@ async fn connect_sftp(
             if let Err(e) = res {
                 // ignore error if dir already exists
                 if !is_sftp_protocol_error(&e) {
-                    return Err(e.into());
+                    return Err(parse_sftp_error(e));
                 }
             }
             fs.set_cwd(&current);
diff --git a/core/src/services/sftp/error.rs b/core/src/services/sftp/error.rs
index 6e08959a5d2a..4c6fd31beaad 100644
--- a/core/src/services/sftp/error.rs
+++ b/core/src/services/sftp/error.rs
@@ -22,55 +22,23 @@ use openssh_sftp_client::Error as SftpClientError;
 use crate::Error;
 use crate::ErrorKind;
 
-#[derive(Debug)]
-pub enum SftpError {
-    SftpClientError(SftpClientError),
-    SshError(SshError),
-}
-
-impl From<SftpClientError> for Error {
-    fn from(e: SftpClientError) -> Self {
-        let kind = match &e {
-            SftpClientError::UnsupportedSftpProtocol { version: _ } => ErrorKind::Unsupported,
-            SftpClientError::SftpError(kind, _msg) => match kind {
-                SftpErrorKind::NoSuchFile => ErrorKind::NotFound,
-                SftpErrorKind::PermDenied => ErrorKind::PermissionDenied,
-                SftpErrorKind::OpUnsupported => ErrorKind::Unsupported,
-                _ => ErrorKind::Unexpected,
-            },
+pub fn parse_sftp_error(e: SftpClientError) -> Error {
+    let kind = match &e {
+        SftpClientError::UnsupportedSftpProtocol { version: _ } => ErrorKind::Unsupported,
+        SftpClientError::SftpError(kind, _msg) => match kind {
+            SftpErrorKind::NoSuchFile => ErrorKind::NotFound,
+            SftpErrorKind::PermDenied => ErrorKind::PermissionDenied,
+            SftpErrorKind::OpUnsupported => ErrorKind::Unsupported,
             _ => ErrorKind::Unexpected,
-        };
-
-        Error::new(kind, "sftp error").set_source(e)
-    }
-}
-
-/// REMOVE ME: it's not allowed to impl `<T>` for Error.
-impl From<SshError> for Error {
-    fn from(e: SshError) -> Self {
-        Error::new(ErrorKind::Unexpected, "ssh error").set_source(e)
-    }
-}
-
-impl From<SftpClientError> for SftpError {
-    fn from(e: SftpClientError) -> Self {
-        SftpError::SftpClientError(e)
-    }
-}
+        },
+        _ => ErrorKind::Unexpected,
+    };
 
-impl From<SshError> for SftpError {
-    fn from(e: SshError) -> Self {
-        SftpError::SshError(e)
-    }
+    Error::new(kind, "sftp error").set_source(e)
 }
 
-impl From<SftpError> for Error {
-    fn from(e: SftpError) -> Self {
-        match e {
-            SftpError::SftpClientError(e) => e.into(),
-            SftpError::SshError(e) => e.into(),
-        }
-    }
+pub fn parse_ssh_error(e: SshError) -> Error {
+    Error::new(ErrorKind::Unexpected, "ssh error").set_source(e)
 }
 
 pub(super) fn is_not_found(e: &SftpClientError) -> bool {
diff --git a/core/src/services/sftp/lister.rs b/core/src/services/sftp/lister.rs
index 14d0b4843f8c..6192fa3fccdf 100644
--- a/core/src/services/sftp/lister.rs
+++ b/core/src/services/sftp/lister.rs
@@ -28,6 +28,8 @@ use openssh_sftp_client::fs::ReadDir;
 use crate::raw::oio;
 use crate::Result;
 
+use super::error::parse_sftp_error;
+
 pub struct SftpLister {
     dir: Pin<Box<ReadDir>>,
     prefix: String,
@@ -47,7 +49,9 @@ impl SftpLister {
 #[async_trait]
 impl oio::List for SftpLister {
     fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
-        let item = ready!(self.dir.poll_next_unpin(cx)).transpose()?;
+        let item = ready!(self.dir.poll_next_unpin(cx))
+            .transpose()
+            .map_err(parse_sftp_error)?;
 
         match item {
             Some(e) => {