Repository: spotify/luigi
Pull request: Added optional choice for hdfs clients 
## Description
Added optional choice for hdfs clients. Previously only InsecureClient was supported. This PR adds KerberosClient and TokenClient. 

## Motivation and Context
Some hdfs work with kerberos authentication. Mine does... This adds support for those clusters. 

This fixes #2481.

## Have you tested this? If so, how?

I have tested the KerberosClient in practice and it works! (Thousand times faster than hadoopcli :) ) 

Not tested the TokenClient though, because I do not have access to a Hadoop cluster with token authentication. 

Problem statement:
Use of KerberosClient with webhdfs
Currently webhdfs uses InsecureClient by default. However, the hdfs package also has an KerberosClient which can be used to do authenticated requests. 

Our hdfs system uses Kerberos and I would like it if KerberosClient is added as an option. 

The KerberosClient does not need any additional settings (as far as I can test). And thus only an additional parameter is needed: `use_kerberos_client`. 

Are you willing to accept a pull request for this?

Hints from issue discussion:
I, personally, do not use the HDFS/Hadoop side of Luigi. But so long as your suggested PR doesn't change the default behavior, please feel free to submit for review!
> Are you willing to accept a pull request for this?

Yep, for sure.

Commit message: Added documentation for the new client_type and token parameter of the webhdfs config documentation section

Patch to review:
diff --git a/doc/configuration.rst b/doc/configuration.rst
index dec55c3679..7305d1756d 100644
--- a/doc/configuration.rst
+++ b/doc/configuration.rst
@@ -922,11 +922,19 @@ summary-length
 port
   The port to use for webhdfs. The normal namenode port is probably on a
   different port from this one.
+
 user
   Perform file system operations as the specified user instead of $USER.  Since
   this parameter is not honored by any of the other hdfs clients, you should
   think twice before setting this parameter.
 
+client_type
+  The type of client to use. Default is the InsecureClient that requires no
+  authentication. Other hdfs clients are KerberosClient and TokenClient.
+
+token
+  The Hadoop delagation token. Only used when client_type is TokenClient.
+
 
 Per Task Retry-Policy
 ---------------------
diff --git a/luigi/contrib/hdfs/webhdfs_client.py b/luigi/contrib/hdfs/webhdfs_client.py
index b350eb79e2..b6ef426bd0 100644
--- a/luigi/contrib/hdfs/webhdfs_client.py
+++ b/luigi/contrib/hdfs/webhdfs_client.py
@@ -43,6 +43,9 @@ class webhdfs(luigi.Config):
                               description='Port for webhdfs')
     user = luigi.Parameter(default='', description='Defaults to $USER envvar',
                            config_path=dict(section='hdfs', name='user'))
+    client_type = luigi.Parameter(default='insecure',
+                                  description='Type of client to use. One of insecure, kerberos or token')
+    token = luigi.Parameter(default='', description='Hadoop delegation token, only used when client_type="token"')
 
 
 class WebHdfsClient(hdfs_abstract_client.HdfsFileSystem):
@@ -53,10 +56,12 @@ class WebHdfsClient(hdfs_abstract_client.HdfsFileSystem):
     <https://hdfscli.readthedocs.io/en/latest/api.html>`__.
     """
 
-    def __init__(self, host=None, port=None, user=None):
+    def __init__(self, host=None, port=None, user=None, client_type=None, token=None):
         self.host = host or hdfs_config.hdfs().namenode_host
         self.port = port or webhdfs().port
         self.user = user or webhdfs().user or os.environ['USER']
+        self.client_type = client_type or webhdfs().client_type
+        self.token = token or webhdfs().token
 
     @property
     def url(self):
@@ -73,8 +78,18 @@ def client(self):
         # not urgent to memoize it. Note that it *might* be issues with process
         # forking and whatnot (as the one in the snakebite client) if we
         # memoize it too trivially.
-        import hdfs
-        return hdfs.InsecureClient(url=self.url, user=self.user)
+        if self.client_type == 'insecure':
+            import hdfs
+            return hdfs.InsecureClient(url=self.url, user=self.user)
+        elif self.client_type == 'kerberos':
+            from hdfs.ext.kerberos import KerberosClient
+            return KerberosClient(url=self.url)
+        elif self.client_type == 'token':
+            import hdfs
+            return hdfs.TokenClient(url=self.url, token=self.token)
+        else:
+            raise ValueError("Error: Unknown client type specified in webhdfs client_type"
+                             "configuration parameter")
 
     def walk(self, path, depth=1):
         return self.client.walk(path, depth=depth)


Code review comments:
diff_hunk: @@ -43,6 +43,9 @@ class webhdfs(luigi.Config):
                               description='Port for webhdfs')
     user = luigi.Parameter(default='', description='Defaults to $USER envvar',
                            config_path=dict(section='hdfs', name='user'))
+    client_type = luigi.Parameter(default='insecure',
path: luigi/contrib/hdfs/webhdfs_client.py
text: @user1:
Maybe use ChoiceParameter?

diff_hunk: @@ -73,8 +78,18 @@ def client(self):
         # not urgent to memoize it. Note that it *might* be issues with process
         # forking and whatnot (as the one in the snakebite client) if we
         # memoize it too trivially.
-        import hdfs
-        return hdfs.InsecureClient(url=self.url, user=self.user)
+        if self.client_type == 'insecure':
+            import hdfs
+            return hdfs.InsecureClient(url=self.url, user=self.user)
+        elif self.client_type == 'kerberos':
+            from hdfs.ext.kerberos import KerberosClient
+            return KerberosClient(url=self.url)
+        elif self.client_type == 'token':
+            import hdfs
+            return hdfs.TokenClient(url=self.url, token=self.token)
+        else:
+            raise ValueError("Error: Unknown client type specified in webhdfs client_type"
path: luigi/contrib/hdfs/webhdfs_client.py
text: @user1:
You could skip this if you use the ChoiceParameter.