Add kerberos keytab authorization

README.md

@@ -28,6 +28,10 @@ A File Output Plugin for Embulk to write HDFS.
     - `RECURSIVE`: delete files and directories
 - **mode**: "abort_if_exist", "overwrite", "delete_files_in_advance", "delete_recursive_in_advance", or "replace". See below. (string, optional, default: `"abort_if_exist"`)
     * In the future, default mode will become `"replace"`.
+- **keytab_config**: For keytab auth for kerberos. see https://docs.cloudera.com/documentation/enterprise/6/6.2/topics/cdh_sg_princ_auth_java.html (hash, default: `{}`)
+  - **krb5_config_path**: krb5.conf file path (string, optional)
+  - **keytab_principal**: user principal (string, optional)
+  - **keytab_path**: Keytab file path (string, optional)
 
 ## CAUTION
 If you use `hadoop` user (hdfs admin user) as `doas`, and if `delete_in_advance` is `RECURSIVE`,

build.gradle

@@ -15,7 +15,7 @@ configurations {
     provided
 }
 
-version = "0.4.0.pre"
+version = "0.4.2.pre"
 
 sourceCompatibility = 1.8
 targetCompatibility = 1.8

src/main/java/org/embulk/output/hdfs/HdfsFileOutputPlugin.java

@@ -59,6 +59,9 @@ public interface PluginTask
         @ConfigDefault("null")
         Optional<String> getDoas();
 
+        @Config("keytab_config")
+        @ConfigDefault("{}")
+        Map<String, String> getKeytabConfig();
         @Deprecated
         enum DeleteInAdvancePolicy
         {

src/main/java/org/embulk/output/hdfs/client/HdfsClient.java

@@ -1,6 +1,7 @@
 package org.embulk.output.hdfs.client;
 
 import com.google.common.base.Optional;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileContext;
 import org.apache.hadoop.fs.FileStatus;
@@ -17,6 +18,7 @@
 import org.slf4j.Logger;
 
 import java.io.File;
+import java.io.IOException;
 import java.io.OutputStream;
 import java.net.MalformedURLException;
 import java.net.URI;
@@ -28,11 +30,44 @@ public class HdfsClient
     public static HdfsClient build(HdfsFileOutputPlugin.PluginTask task)
     {
         Configuration conf = buildConfiguration(task.getConfigFiles(), task.getConfig());
+        setKerberosKeytabAuthention(conf, task.getKeytabConfig());
         return new HdfsClient(conf, task.getDoas());
     }
 
     ;
 
+    /**
+     * https://docs.cloudera.com/documentation/enterprise/6/6.2/topics/cdh_sg_princ_auth_java.html
+     */
+    public static void setKerberosKeytabAuthention(Configuration conf, Map<String, String> keytabConfig){
+        if(keytabConfig == null || keytabConfig.size() == 0){
+            UserGroupInformation.setConfiguration(conf);
+            return;
+        }
+        String krb5ConfigPath = keytabConfig.get("krb5_config_path");
+        String keytabPrincipal = keytabConfig.get("keytab_principal");
+        String keytabPath = keytabConfig.get("keytab_path");
+
+        logger.info("Keytab config init. krb5_config_path: {} , keytab_principal : {}, keytab_path : {}", krb5ConfigPath, keytabPrincipal, keytabPath);
+
+        if(StringUtils.isEmpty(krb5ConfigPath) ||
+                StringUtils.isEmpty(keytabPrincipal) ||
+                StringUtils.isEmpty(keytabPath)){
+            throw new ConfigException(String.format("Keytab config not enough. krb5_config_path: {} , keytab_principal : {}, keytab_path : {}", krb5ConfigPath, keytabPrincipal, keytabPath));
+        }
+
+        try {
+            System.setProperty("java.security.krb5.conf", krb5ConfigPath);
+            UserGroupInformation.setConfiguration(conf);
+            UserGroupInformation.loginUserFromKeytab(keytabPrincipal, keytabPath);
+        } catch (IOException e) {
+            throw new ConfigException(e);
+        }
+
+    }
+
+    ;
+
     public static Configuration buildConfiguration(List<String> configFiles, Map<String, String> configs)
     {
         Configuration c = new Configuration();
@@ -48,7 +83,6 @@ public static Configuration buildConfiguration(List<String> configFiles, Map<Str
         for (Map.Entry<String, String> config : configs.entrySet()) {
             c.set(config.getKey(), config.getValue());
         }
-        UserGroupInformation.setConfiguration(c);
         return c;
     }