From 39a204b4994e8448a81f3b403e63716d227263e2 Mon Sep 17 00:00:00 2001 From: sakama Date: Wed, 28 Feb 2018 18:23:30 +0900 Subject: [PATCH 1/2] Fix SFTP connection remaining problem when listing remote files --- .../org/embulk/input/sftp/SftpFileInput.java | 64 ++++++++++--------- 1 file changed, 35 insertions(+), 29 deletions(-) diff --git a/src/main/java/org/embulk/input/sftp/SftpFileInput.java b/src/main/java/org/embulk/input/sftp/SftpFileInput.java index 93eaf58..6375a6f 100644 --- a/src/main/java/org/embulk/input/sftp/SftpFileInput.java +++ b/src/main/java/org/embulk/input/sftp/SftpFileInput.java @@ -189,43 +189,49 @@ public FileList call() throws IOException { String lastKey = null; log.info("Getting to download file list"); - StandardFileSystemManager manager = initializeStandardFileSystemManager(); - FileSystemOptions fsOptions = initializeFsOptions(task); + StandardFileSystemManager manager = null; + try { + manager = initializeStandardFileSystemManager(); + FileSystemOptions fsOptions = initializeFsOptions(task); - if (task.getLastPath().isPresent() && !task.getLastPath().get().isEmpty()) { - lastKey = manager.resolveFile(getSftpFileUri(task, task.getLastPath().get()), fsOptions).toString(); - } + if (task.getLastPath().isPresent() && !task.getLastPath().get().isEmpty()) { + lastKey = manager.resolveFile(getSftpFileUri(task, task.getLastPath().get()), fsOptions).toString(); + } - FileObject files = manager.resolveFile(getSftpFileUri(task, task.getPathPrefix()), fsOptions); + FileObject files = manager.resolveFile(getSftpFileUri(task, task.getPathPrefix()), fsOptions); - if (files.isFolder()) { - //path_prefix is a folder, we add everything in that folder - FileObject[] children = files.getChildren(); - Arrays.sort(children); - for (FileObject f : children) { - if (f.isFile()) { - addFileToList(builder, f.toString(), f.getContent().getSize(), "", lastKey); + if (files.isFolder()) { + //path_prefix is a folder, we add everything in that folder + FileObject[] children = files.getChildren(); + Arrays.sort(children); + for (FileObject f : children) { + if (f.isFile()) { + addFileToList(builder, f.toString(), f.getContent().getSize(), "", lastKey); + } + } + } else if (files.isFile()) { + //path_prefix is a file then we just need to add that file + addFileToList(builder, files.toString(), files.getContent().getSize(), "", lastKey); + } else { + // path_prefix is neither file or folder, then we scan the parent folder to file path + // that match the path_prefix basename + FileObject parent = files.getParent(); + FileObject[] children = parent.getChildren(); + Arrays.sort(children); + String fileName = FilenameUtils.getName(task.getPathPrefix()); + for (FileObject f : children) { + if (f.isFile()) { + addFileToList(builder, f.toString(), f.getContent().getSize(), fileName, lastKey); + } } } + return builder.build(); } - else if (files.isFile()) { - //path_prefix is a file then we just need to add that file - addFileToList(builder, files.toString(), files.getContent().getSize(), "", lastKey); - } - else { - // path_prefix is neither file or folder, then we scan the parent folder to file path - // that match the path_prefix basename - FileObject parent = files.getParent(); - FileObject[] children = parent.getChildren(); - Arrays.sort(children); - String fileName = FilenameUtils.getName(task.getPathPrefix()); - for (FileObject f : children) { - if (f.isFile()) { - addFileToList(builder, f.toString(), f.getContent().getSize(), fileName, lastKey); - } + finally { + if (manager != null) { + manager.close(); } } - return builder.build(); } @Override From dc5f3f24a371cf239dc622f941cbd216928d0073 Mon Sep 17 00:00:00 2001 From: sakama Date: Thu, 1 Mar 2018 13:37:25 +0900 Subject: [PATCH 2/2] Call StandardFileSystemManager.close() in SingleFileProvider.close() --- src/main/java/org/embulk/input/sftp/SftpFileInput.java | 1 + src/main/java/org/embulk/input/sftp/SingleFileProvider.java | 3 +++ 2 files changed, 4 insertions(+) diff --git a/src/main/java/org/embulk/input/sftp/SftpFileInput.java b/src/main/java/org/embulk/input/sftp/SftpFileInput.java index 6375a6f..2f04417 100644 --- a/src/main/java/org/embulk/input/sftp/SftpFileInput.java +++ b/src/main/java/org/embulk/input/sftp/SftpFileInput.java @@ -56,6 +56,7 @@ public TaskReport commit() @Override public void close() { + super.close(); } private static StandardFileSystemManager initializeStandardFileSystemManager() diff --git a/src/main/java/org/embulk/input/sftp/SingleFileProvider.java b/src/main/java/org/embulk/input/sftp/SingleFileProvider.java index a563721..ce208ed 100644 --- a/src/main/java/org/embulk/input/sftp/SingleFileProvider.java +++ b/src/main/java/org/embulk/input/sftp/SingleFileProvider.java @@ -99,5 +99,8 @@ public void onGiveup(Exception firstException, Exception lastException) @Override public void close() { + if (manager != null) { + manager.close(); + } } }