Implement avoid_create_0byte_files

Author: civitaspo

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

@@ -122,30 +122,28 @@ public TransactionalFileOutput open(TaskSource taskSource, final int taskIndex)
         {
             private final List<String> hdfsFileNames = new ArrayList<>();
             private int fileIndex = 0;
+            private Path currentPath = null;
             private OutputStream output = null;
 
             @Override
             public void nextFile()
             {
                 closeCurrentStream();
-                Path path = new Path(pathPrefix + String.format(sequenceFormat, taskIndex, fileIndex) + pathSuffix);
-                try {
-                    FileSystem fs = getFs(task);
-                    output = fs.create(path, task.getOverwrite());
-                    logger.info("Uploading '{}'", path);
-                }
-                catch (IOException e) {
-                    logger.error(e.getMessage());
-                    throw new RuntimeException(e);
-                }
-                hdfsFileNames.add(path.toString());
+                currentPath = new Path(pathPrefix + String.format(sequenceFormat, taskIndex, fileIndex) + pathSuffix);
                 fileIndex++;
             }
 
             @Override
             public void add(Buffer buffer)
             {
                 try {
+                    // this implementation is for creating file when there is data.
+                    if (output == null) {
+                        FileSystem fs = getFs(task);
+                        output = fs.create(currentPath, task.getOverwrite());
+                        logger.info("Uploading '{}'", currentPath);
+                        hdfsFileNames.add(currentPath.toString());
+                    }
                     output.write(buffer.array(), buffer.offset(), buffer.limit());
                 }
                 catch (IOException e) {