Merge pull request #10 from civitaspo/v0.2.2

V0.2.2

Author: civitaspo

CHANGELOG.md

@@ -0,0 +1,3 @@
+0.2.2 (2016-02-02)
+==================
+- Add: doas option

build.gradle

@@ -12,7 +12,7 @@ configurations {
     provided
 }
 
-version = "0.2.1"
+version = "0.2.2"
 
 sourceCompatibility = 1.7
 targetCompatibility = 1.7

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

@@ -1,21 +1,16 @@
 package org.embulk.output.hdfs;
 
-import java.io.File;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
+import com.google.common.base.Optional;
+import com.google.common.base.Throwables;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.embulk.config.TaskReport;
 import org.embulk.config.Config;
 import org.embulk.config.ConfigDefault;
 import org.embulk.config.ConfigDiff;
 import org.embulk.config.ConfigSource;
 import org.embulk.config.Task;
+import org.embulk.config.TaskReport;
 import org.embulk.config.TaskSource;
 import org.embulk.spi.Buffer;
 import org.embulk.spi.Exec;
@@ -24,6 +19,14 @@
 import org.jruby.embed.ScriptingContainer;
 import org.slf4j.Logger;
 
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
 public class HdfsFileOutputPlugin
         implements FileOutputPlugin
 {
@@ -58,6 +61,9 @@ public interface PluginTask
         @ConfigDefault("false")
         public boolean getOverwrite();
 
+        @Config("doas")
+        @ConfigDefault("null")
+        public Optional<String> getDoas();
     }
 
     @Override
@@ -186,6 +192,15 @@ private static FileSystem getFs(final PluginTask task)
             configuration.set(entry.getKey(), entry.getValue());
         }
 
+        if (task.getDoas().isPresent()) {
+            URI uri = FileSystem.getDefaultUri(configuration);
+            try {
+                return FileSystem.get(uri, configuration, task.getDoas().get());
+            }
+            catch (InterruptedException e) {
+                throw Throwables.propagate(e);
+            }
+        }
         return FileSystem.get(configuration);
     }