fix a bug: Could not write to "ram:///tmp/hoge" because it is currently in use.

Author: civitaspo

src/main/java/org/embulk/output/sftp/SftpFileOutput.java

@@ -42,6 +42,7 @@ public class SftpFileOutput
     private final int taskIndex;
     private int fileIndex = 0;
     private FileObject currentRamFile;
+    private OutputStream currentRamFileOutputStream;
 
     private StandardFileSystemManager initializeStandardFileSystemManager()
     {
@@ -109,6 +110,7 @@ public void nextFile()
 
         try {
             currentRamFile = newRamFile(getRamUri(getOutputFilePath()));
+            currentRamFileOutputStream = getCurrentRamFileOutputStream();
             logger.info("new ram file: {}", currentRamFile.getPublicURIString());
         }
         catch (FileSystemException e) {
@@ -129,7 +131,7 @@ public void add(Buffer buffer)
         }
 
         try {
-            getCurrentRamFileOutputStream().write(buffer.array(), buffer.offset(), buffer.limit());
+            currentRamFileOutputStream.write(buffer.array(), buffer.offset(), buffer.limit());
         }
         catch (IOException e) {
             logger.error(e.getMessage());
@@ -169,11 +171,11 @@ private void closeCurrentWithUpload()
             uploadCurrentRamFileToSftp();
             closeCurrentRamFile();
         }
-        catch (FileSystemException e) {
+        catch (URISyntaxException e) {
             logger.error(e.getMessage());
             Throwables.propagate(e);
         }
-        catch (URISyntaxException e) {
+        catch (IOException e) {
             logger.error(e.getMessage());
             Throwables.propagate(e);
         }
@@ -182,11 +184,12 @@ private void closeCurrentWithUpload()
     }
 
     private void closeCurrentRamFileContent()
-            throws FileSystemException
+            throws IOException
     {
         if (currentRamFile == null) {
             return;
         }
+        currentRamFileOutputStream.close();
         currentRamFile.getContent().close();
     }