Add `local_buffering` & `temp_file_threshold` configs #50

tvhung83 · 2018-08-14T14:25:56Z

CHANGELOG

This PR introduces 2 new configs:

local_buffering: whether to use local temp file as buffer. If false (new behavior), plugin will use remote file as buffering, ie. writing directly to remote OutputStream. This new behavior can help to reduce processing time, and not to use local file, which will reduce usage of disk space.
temp_file_threshold: long, maximum file size of local temp file, before plugin will flush to server.

Note

To implement local_buffering, SftpFileOutput is now SftpLocalFileOutput and SftpRemoteFileOutput:

SftpLocalFileOutput keeps buffering records into local temp file, and only flush to remote file when its size reaches temp_file_threshold.
SftpRemoteFileOutput will use remote file as buffer, hence there is no threshold. The progress will show number of uploaded records, instead of file size.

I also added retry into some operations of SftpUtils.

…terruptible) and callNonInterruptible

…`deleteFile()` method, `withRetry()` utility function

…5 minutes, add `resolve` and `openStream` (with retry)

…en no exception is thrown

sakama

Basically, LGTM
Put some minor comments.

sakama · 2018-08-24T02:47:58Z

build.gradle

-    compile  "org.embulk:embulk-core:0.8.6"
-    provided "org.embulk:embulk-core:0.8.6"
+    compile  "org.embulk:embulk-core:0.8.+"
+    provided "org.embulk:embulk-core:0.8.+"


Fixed version might better and embulk v0.9.x is ok for now?

Even though I'm not very confident with 0.9.x, but I can try that.

sakama · 2018-08-24T02:55:02Z

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

+        // if `false`, plugin will use remote file as buffer
+        @Config("local_buffering")
+        @ConfigDefault("true")
+        public Boolean getLocalBuffering();


✅ will update.

sakama · 2018-08-24T02:55:39Z

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

@@ -74,6 +77,17 @@
        @Config("rename_file_after_upload")
        @ConfigDefault("false")
        public Boolean getRenameFileAfterUpload();


sakama · 2018-08-24T03:21:01Z

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

+            public Void call() throws Exception
+            {
+                final FileObject remoteFile = newSftpFile(getSftpFileUri(remotePath));
+                final OutputStream outputStream = openStream(remoteFile);


Minor: BufferedOutputStream is correct?

The instance is already BufferedOutputStream, and here I only need to use basic operations of OutputStream, I'd like to keep it simple, but I can change it to make it explicit.

sakama · 2018-08-24T03:24:02Z

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

+        long startTime = System.nanoTime();
+
+        // start uploading
+        try (InputStream inputStream = new FileInputStream(localTempFile)) {


Although original implementation was wrong, Use try (BufferedInputStream inputStream = new BufferedInputStream(new FileInputStream(localTempFile)) {} is better?

Good catch, will fix it, thanks.

sakama · 2018-08-24T03:27:59Z

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

+     * @param outputStream
+     * @throws IOException
+     */
+    void appendFile(final File localTempFile, final FileObject remoteFile, final OutputStream outputStream) throws IOException


Minor: BurreredOutputStream outputStream is correct?
I think you already passed BurreredOutputStream

Will update.

sakama · 2018-08-24T03:43:45Z

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

+    }
+
+    @VisibleForTesting
+    OutputStream getRemoteOutput()


Minor: BufferedOutputStream is correct?

It's just for testing, can I keep it that way?

tvhung83 · 2018-08-24T05:40:04Z

Checking unit tests...

sakama

LGTM

tvhung83 · 2018-08-24T07:21:49Z

@sakama -san Please help take another look: tvhung83/embulk-output-sftp@bc94c9f...8b8ca77
After more tests, I found out that it's not safe to ignore TimeoutException in TimeoutCloser. When the connection is stalling (close output stream → flush → stalled), the remote file would be broken. Below are what changed:

Revert input stream back to non-buffered stream, because we already read by buffer, 2 level of buffers is unpredictable.
Abort when closing resource got timed out.
Increase default time-out duration to 5 minutes (instead of 2 minutes).
Remove obsolete method callNonInterruptible()

sakama

LGTM

tvhung83 · 2018-08-27T07:03:49Z

Thanks, going to merge now.

tvhung83 added 30 commits August 14, 2018 21:17

Add utility class: default as always retry

633df74

A time-out Callable implementation, which exposes 2 methods: call (in…

40372f0

…terruptible) and callNonInterruptible

Rename configs

5beb673

Add flush() and abort() implementation

a3e3eeb

Rename TMP_SUFFIX constant

9d7eabb

Adjust uploadFile() method: non-retry to preserve idempotence, add …

8d5a8a3

…`deleteFile()` method, `withRetry()` utility function

Not retry for Auth error

344d4cd

Revert to 5 retries limit

5ff77dc

Fix some logic mistakes: closing resources, rename in finish, etc.

3e4ba17

Add progress watcher and overwrite add, closeCurrentFile

8f6edd4

Separate uploadFile and appendFile, increase default time-out to …

df0fa17

…5 minutes, add `resolve` and `openStream` (with retry)

Fix a bug of task.cancel(), which keeps the task running forever wh…

d5057a7

…en no exception is thrown

A wrapper class to close resource and time-out if needed

8f664a5

Default local temp file threshold to 5GiB

ac8794c

Refactor: rearrange in attempt to make workflow cleaner

c78538d

Fix bug: close/flush final file, move rename right after closing file

f6623ba

Update code comments

aa0f97c

Reset appending when open next file

ba26217

Increase time-out to 2 minutes

2ee2a04

Switch to BufferedOutputStream for remote output stream

f6e49d1

Fix bug: remote file was not flushed properly

3c9d9eb

Fix bug: close remote output stream instead of file

300df83

Revert name of test class, remove JDK7 from .travis.yml

03e1eec

Upgrade dist to trusty in attempt to make openjdk8 works

590d967

Update embulk-* to latest 0.8, add mockito for unit test

221ac5e

Fix checkstyle

c43da7d

Minor bug fix, open some methods for unit tests

e43e18f

Add more unit tests

d563378

Minor bug fix, open writeTimeout for unit tests

0b3bc4f

Open timeout for testing

82356c7

tvhung83 changed the title ~~[WIP] Add local_buffering & temp_file_threshold configs~~ Add local_buffering & temp_file_threshold configs Aug 23, 2018

sakama self-requested a review August 24, 2018 02:32

tvhung83 added 2 commits August 24, 2018 11:09

Skip closing remoteFile because it hangs jobs

1e62158

Remove retry for resolve since it's causing nested retries

a8bb85b

sakama reviewed Aug 24, 2018

View reviewed changes

tvhung83 added 2 commits August 24, 2018 12:07

Address review comments, also removed an obsolete test

37e50b6

Update embulk-* to 0.9.7

5c24bf4

Fix bug: move renaming to finish() instead of closeRemoteFile()

bc94c9f

sakama approved these changes Aug 24, 2018

View reviewed changes

tvhung83 added 4 commits August 24, 2018 14:14

Revert to non-buffered input stream, because we already read into buffer

85241a4

Remove obsolete callNonInterruptible() method

4ae6231

Remove obsolete callNonInterruptible() method

70a5429

Update unit test of time-out closer

7ebdd87

Remove unnecessary close (it could cause another hang)

8b8ca77

sakama approved these changes Aug 27, 2018

View reviewed changes

tvhung83 merged commit 6b61c6e into embulk:master Aug 27, 2018

tvhung83 deleted the add_local_buffering_config branch August 27, 2018 07:04

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add `local_buffering` & `temp_file_threshold` configs #50

Add `local_buffering` & `temp_file_threshold` configs #50

tvhung83 commented Aug 14, 2018 •

edited

Loading

sakama left a comment

sakama Aug 24, 2018

tvhung83 Aug 24, 2018

sakama Aug 24, 2018

tvhung83 Aug 24, 2018

sakama Aug 24, 2018

tvhung83 Aug 24, 2018

sakama Aug 24, 2018

tvhung83 Aug 24, 2018

sakama Aug 24, 2018

tvhung83 Aug 24, 2018

sakama Aug 24, 2018

tvhung83 Aug 24, 2018

sakama Aug 24, 2018

tvhung83 Aug 24, 2018

sakama Aug 24, 2018

tvhung83 commented Aug 24, 2018

sakama left a comment

tvhung83 commented Aug 24, 2018 •

edited

Loading

sakama left a comment

tvhung83 commented Aug 27, 2018

Add local_buffering & temp_file_threshold configs #50

Add local_buffering & temp_file_threshold configs #50

Conversation

tvhung83 commented Aug 14, 2018 • edited Loading

CHANGELOG

Note

sakama left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tvhung83 commented Aug 24, 2018

sakama left a comment

Choose a reason for hiding this comment

tvhung83 commented Aug 24, 2018 • edited Loading

sakama left a comment

Choose a reason for hiding this comment

tvhung83 commented Aug 27, 2018

Add `local_buffering` & `temp_file_threshold` configs #50

Add `local_buffering` & `temp_file_threshold` configs #50

tvhung83 commented Aug 14, 2018 •

edited

Loading

tvhung83 commented Aug 24, 2018 •

edited

Loading