ARROW-350: Added Kerberos to HDFS client

cpp/src/arrow/io/hdfs.cc

@@ -287,12 +287,25 @@ class HdfsClient::HdfsClientImpl {
   Status Connect(const HdfsConnectionConfig* config) {
     RETURN_NOT_OK(ConnectLibHdfs());
 
-    fs_ = hdfsConnectAsUser(config->host.c_str(), config->port, config->user.c_str());
+    // connect to HDFS with the builder object
+    hdfsBuilder* builder = hdfsNewBuilder();
+    if (!config->host.empty()) {
+      hdfsBuilderSetNameNode(builder, config->host.c_str());
+    }
+    hdfsBuilderSetNameNodePort(builder, config->port);
+    if (!config->user.empty()) {
+      hdfsBuilderSetUserName(builder, config->user.c_str());
+    }
+    if (!config->kerb.empty()) {
+      hdfsBuilderSetKerbTicketCachePath(builder, config->kerb.c_str());
+    }
+    fs_ = hdfsBuilderConnect(builder);
 
     if (fs_ == nullptr) { return Status::IOError("HDFS connection failed"); }
     namenode_host_ = config->host;
     port_ = config->port;
     user_ = config->user;
+    kerb_ = config->kerb;
 
     return Status::OK();
   }
@@ -425,6 +438,7 @@ class HdfsClient::HdfsClientImpl {
   std::string namenode_host_;
   std::string user_;
   int port_;
+  std::string kerb_;
 
   hdfsFS fs_;
 };

cpp/src/arrow/io/hdfs.h

@@ -60,19 +60,16 @@ struct HdfsConnectionConfig {
   std::string host;
   int port;
   std::string user;
-
-  // TODO: Kerberos, etc.
+  std::string kerb;
 };
 
 class ARROW_EXPORT HdfsClient : public FileSystemClient {
  public:
   ~HdfsClient();
 
-  // Connect to an HDFS cluster at indicated host, port, and as user
+  // Connect to an HDFS cluster given a configuration
   //
-  // @param host (in)
-  // @param port (in)
-  // @param user (in): user to identify as
+  // @param config (in): configuration for connecting
   // @param fs (out): the created client
   // @returns Status
   static Status Connect(

cpp/src/arrow/io/libhdfs_shim.cc

@@ -73,9 +73,17 @@ static HINSTANCE libjvm_handle = NULL;
 
 // NOTE(wesm): cpplint does not like use of short and other imprecise C types
 
-static hdfsFS (*ptr_hdfsConnectAsUser)(
-    const char* host, tPort port, const char* user) = NULL;
-static hdfsFS (*ptr_hdfsConnect)(const char* host, tPort port) = NULL;
+static hdfsBuilder* (*ptr_hdfsNewBuilder)(void) = NULL;
+static void (*ptr_hdfsBuilderSetNameNode)(
+    hdfsBuilder* bld, const char* nn) = NULL;
+static void (*ptr_hdfsBuilderSetNameNodePort)(
+    hdfsBuilder* bld, tPort port) = NULL;
+static void (*ptr_hdfsBuilderSetUserName)(
+    hdfsBuilder* bld, const char* userName) = NULL;
+static void (*ptr_hdfsBuilderSetKerbTicketCachePath)(
+    hdfsBuilder* bld, const char* kerbTicketCachePath) = NULL;
+static hdfsFS (*ptr_hdfsBuilderConnect)(hdfsBuilder* bld) = NULL;
+
 static int (*ptr_hdfsDisconnect)(hdfsFS fs) = NULL;
 
 static hdfsFile (*ptr_hdfsOpenFile)(hdfsFS fs, const char* path, int flags,
@@ -149,18 +157,29 @@ static void* get_symbol(const char* symbol) {
 #endif
 }
 
-hdfsFS hdfsConnectAsUser(const char* host, tPort port, const char* user) {
-  return ptr_hdfsConnectAsUser(host, port, user);
+hdfsBuilder* hdfsNewBuilder(void) {
+  return ptr_hdfsNewBuilder();
 }
 
-// Returns NULL on failure
-hdfsFS hdfsConnect(const char* host, tPort port) {
-  if (ptr_hdfsConnect) {
-    return ptr_hdfsConnect(host, port);
-  } else {
-    // TODO: error reporting when shim setup fails
-    return NULL;
-  }
+void hdfsBuilderSetNameNode(hdfsBuilder* bld, const char* nn) {
+  ptr_hdfsBuilderSetNameNode(bld, nn);
+}
+
+void hdfsBuilderSetNameNodePort(hdfsBuilder* bld, tPort port) {
+  ptr_hdfsBuilderSetNameNodePort(bld, port);
+}
+
+void hdfsBuilderSetUserName(hdfsBuilder* bld, const char* userName) {
+  ptr_hdfsBuilderSetUserName(bld, userName);
+}
+
+void hdfsBuilderSetKerbTicketCachePath(hdfsBuilder* bld,
+    const char* kerbTicketCachePath) {
+  ptr_hdfsBuilderSetKerbTicketCachePath(bld , kerbTicketCachePath);
+}
+
+hdfsFS hdfsBuilderConnect(hdfsBuilder* bld) {
+  return ptr_hdfsBuilderConnect(bld);
 }
 
 int hdfsDisconnect(hdfsFS fs) {
@@ -531,8 +550,12 @@ Status ARROW_EXPORT ConnectLibHdfs() {
     return Status::IOError("Prior attempt to load libhdfs failed");
   }
 
-  GET_SYMBOL_REQUIRED(hdfsConnect);
-  GET_SYMBOL_REQUIRED(hdfsConnectAsUser);
+  GET_SYMBOL_REQUIRED(hdfsNewBuilder);
+  GET_SYMBOL_REQUIRED(hdfsBuilderSetNameNode);
+  GET_SYMBOL_REQUIRED(hdfsBuilderSetNameNodePort);
+  GET_SYMBOL_REQUIRED(hdfsBuilderSetUserName);
+  GET_SYMBOL_REQUIRED(hdfsBuilderSetKerbTicketCachePath);
+  GET_SYMBOL_REQUIRED(hdfsBuilderConnect);
   GET_SYMBOL_REQUIRED(hdfsCreateDirectory);
   GET_SYMBOL_REQUIRED(hdfsDelete);
   GET_SYMBOL_REQUIRED(hdfsDisconnect);

python/pyarrow/includes/libarrow_io.pxd

@@ -93,6 +93,7 @@ cdef extern from "arrow/io/hdfs.h" namespace "arrow::io" nogil:
         c_string host
         int port
         c_string user
+        c_string kerb
 
     cdef cppclass HdfsPathInfo:
         ObjectType kind;

python/pyarrow/io.pyx

@@ -288,9 +288,6 @@ cdef class HdfsClient:
         shared_ptr[CHdfsClient] client
 
     cdef readonly:
-        object host
-        int port
-        object user
         bint is_open
 
     def __cinit__(self):
@@ -301,6 +298,9 @@ cdef class HdfsClient:
             self.close()
 
     def close(self):
+        """
+        Disconnect from the HDFS cluster
+        """
         self._ensure_client()
         with nogil:
             check_status(self.client.get().Disconnect())
@@ -313,14 +313,21 @@ cdef class HdfsClient:
             raise IOError('HDFS client is closed')
 
     @classmethod
-    def connect(cls, host, port, user):
+    def connect(cls, host="default", port=0, user=None, kerb=None):
         """
+        Connect to an HDFS cluster. All parameters are optional and should
+        only be set if the defaults need to be overridden.
+
+        Authentication should be automatic if the HDFS cluster uses Kerberos.
+        However, if a username is specified, then the ticket cache will likely
+        be required.
 
         Parameters
         ----------
-        host :
-        port :
-        user :
+        host : NameNode. Set to "default" for fs.defaultFS from core-site.xml.
+        port : NameNode's port. Set to 0 for default or logical (HA) nodes.
+        user : Username when connecting to HDFS; None implies login user.
+        kerb : Path to Kerberos ticket cache.
 
         Notes
         -----
@@ -335,9 +342,13 @@ cdef class HdfsClient:
             HdfsClient out = HdfsClient()
             HdfsConnectionConfig conf
 
-        conf.host = tobytes(host)
+        if host is not None:
+            conf.host = tobytes(host)
         conf.port = port
-        conf.user = tobytes(user)
+        if user is not None:
+            conf.user = tobytes(user)
+        if kerb is not None:
+            conf.kerb = tobytes(kerb)
 
         with nogil:
             check_status(CHdfsClient.Connect(&conf, &out.client))