feat: support kerberos for hdfs service

core/Cargo.toml

@@ -193,7 +193,7 @@ dirs = { version = "5.0.1", optional = true }
 flagset = "0.4"
 futures = { version = "0.3", default-features = false, features = ["std"] }
 governor = { version = "0.5", optional = true, features = ["std"] }
-hdrs = { version = "0.2", optional = true, features = ["async_file"] }
+hdrs = { version = "0.3.0", optional = true, features = ["async_file"] }
 http = "0.2.5"
 hyper = "0.14"
 lazy-regex = { version = "2.5.0", optional = true }

core/src/services/hdfs/backend.rs

@@ -38,6 +38,8 @@ use crate::*;
 pub struct HdfsBuilder {
     root: Option<String>,
     name_node: Option<String>,
+    kerberos_ticket_cache_path: Option<String>,
+    user: Option<String>,
 }
 
 impl HdfsBuilder {
@@ -68,6 +70,24 @@ impl HdfsBuilder {
 
         self
     }
+
+    /// Set kerberos_ticket_cache_path of this backend
+    ///
+    /// This should be configured when kerberos is enabled.
+    pub fn kerberos_ticket_cache_path(&mut self, kerberos_ticket_cache_path: &str) -> &mut Self {
+        if !kerberos_ticket_cache_path.is_empty() {
+            self.kerberos_ticket_cache_path = Some(kerberos_ticket_cache_path.to_string())
+        }
+        self
+    }
+
+    /// Set user of this backend
+    pub fn user(&mut self, user: &str) -> &mut Self {
+        if !user.is_empty() {
+            self.user = Some(user.to_string())
+        }
+        self
+    }
 }
 
 impl Builder for HdfsBuilder {
@@ -79,6 +99,9 @@ impl Builder for HdfsBuilder {
 
         map.get("root").map(|v| builder.root(v));
         map.get("name_node").map(|v| builder.name_node(v));
+        map.get("kerberos_ticket_cache_path")
+            .map(|v| builder.kerberos_ticket_cache_path(v));
+        map.get("user").map(|v| builder.user(v));
 
         builder
     }
@@ -97,7 +120,15 @@ impl Builder for HdfsBuilder {
         let root = normalize_root(&self.root.take().unwrap_or_default());
         debug!("backend use root {}", root);
 
-        let client = hdrs::Client::connect(name_node).map_err(parse_io_error)?;
+        let mut builder = hdrs::ClientBuilder::new(name_node);
+        if let Some(ticket_cache_path) = &self.kerberos_ticket_cache_path {
+            builder = builder.with_kerberos_ticket_cache_path(ticket_cache_path.as_str());
+        }
+        if let Some(user) = &self.user {
+            builder = builder.with_user(user.as_str());
+        }
+
+        let client = builder.connect().map_err(parse_io_error)?;
 
         // Create root dir if not exist.
         if let Err(e) = client.metadata(&root) {

core/src/services/hdfs/docs.md

@@ -28,6 +28,8 @@ HDFS support needs to enable feature `services-hdfs`.
 
 - `root`: Set the work dir for backend.
 - `name_node`: Set the name node for backend.
+- `kerberos_ticket_cache_path`: Set the kerberos ticket cache path for backend, this should be gotten by `klist` after `kinit`
+- `user`: Set the user for backend
 
 Refer to [`HdfsBuilder`]'s public API docs for more information.