You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2017/02/22 20:37:45 UTC
[1/2] beam git commit: [BEAM-1465] No natural place to flush/close
resources in FileBasedWriter
Repository: beam
Updated Branches:
refs/heads/master ede77c1b5 -> 41de9301a
[BEAM-1465] No natural place to flush/close resources in FileBasedWriter
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/2b6a08f6
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/2b6a08f6
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/2b6a08f6
Branch: refs/heads/master
Commit: 2b6a08f6cb37a9c51d9ae5a63e6b69a310bb3aae
Parents: ede77c1
Author: Aviem Zur <av...@gmail.com>
Authored: Wed Feb 22 06:26:38 2017 +0200
Committer: Dan Halperin <dh...@google.com>
Committed: Wed Feb 22 12:37:30 2017 -0800
----------------------------------------------------------------------
.../src/main/java/org/apache/beam/sdk/io/AvroIO.java | 2 +-
.../main/java/org/apache/beam/sdk/io/FileBasedSink.java | 11 +++++++++++
.../src/main/java/org/apache/beam/sdk/io/TextIO.java | 11 +++++++----
3 files changed, 19 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/2b6a08f6/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
index 01a4cba..388d9f0 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
@@ -1032,7 +1032,7 @@ public class AvroIO {
}
@Override
- protected void writeFooter() throws Exception {
+ protected void finishWrite() throws Exception {
dataFileWriter.flush();
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/2b6a08f6/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java
index 32b8b4f..e14ba59 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java
@@ -590,6 +590,12 @@ public abstract class FileBasedSink<T> extends Sink<T> {
protected void writeFooter() throws Exception {}
/**
+ * Called after all calls to {@link #writeHeader}, {@link #write} and {@link #writeFooter}.
+ * If any resources opened in the write processes need to be flushed, flush them here.
+ */
+ protected void finishWrite() throws Exception {}
+
+ /**
* Opens the channel.
*/
@Override
@@ -630,6 +636,11 @@ public abstract class FileBasedSink<T> extends Sink<T> {
try (WritableByteChannel theChannel = channel) {
LOG.debug("Writing footer to {}.", filename);
writeFooter();
+ LOG.debug("Finishing write to {}.", filename);
+ finishWrite();
+ if (!channel.isOpen()) {
+ throw new IllegalStateException("Channel should only be closed by its owner: " + channel);
+ }
}
FileResult result = new FileResult(filename);
LOG.debug("Result for bundle {}: {}", this.id, filename);
http://git-wip-us.apache.org/repos/asf/beam/blob/2b6a08f6/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java
index 726411c..86e6989 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java
@@ -1101,15 +1101,18 @@ public class TextIO {
}
@Override
+ public void write(String value) throws Exception {
+ writeLine(value);
+ }
+
+ @Override
protected void writeFooter() throws Exception {
writeIfNotNull(footer);
- // Flush here because there is currently no other natural place to do this. [BEAM-1465]
- out.flush();
}
@Override
- public void write(String value) throws Exception {
- writeLine(value);
+ protected void finishWrite() throws Exception {
+ out.flush();
}
}
}
[2/2] beam git commit: This closes #2067
Posted by dh...@apache.org.
This closes #2067
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/41de9301
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/41de9301
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/41de9301
Branch: refs/heads/master
Commit: 41de9301af367a2dff19255d2d2fe563368af55c
Parents: ede77c1 2b6a08f
Author: Dan Halperin <dh...@google.com>
Authored: Wed Feb 22 12:37:36 2017 -0800
Committer: Dan Halperin <dh...@google.com>
Committed: Wed Feb 22 12:37:36 2017 -0800
----------------------------------------------------------------------
.../src/main/java/org/apache/beam/sdk/io/AvroIO.java | 2 +-
.../main/java/org/apache/beam/sdk/io/FileBasedSink.java | 11 +++++++++++
.../src/main/java/org/apache/beam/sdk/io/TextIO.java | 11 +++++++----
3 files changed, 19 insertions(+), 5 deletions(-)
----------------------------------------------------------------------