beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From m..@apache.org
Subject [1/2] incubator-beam git commit: [BEAM-207] Flink test flake in ReadSourceStreamingITCase
Date Wed, 20 Apr 2016 17:21:42 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master b8951c231 -> d5b1d5135


[BEAM-207] Flink test flake in ReadSourceStreamingITCase


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/dd8bc93e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/dd8bc93e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/dd8bc93e

Branch: refs/heads/master
Commit: dd8bc93ec3104c481a9ea646406194c1116dae71
Parents: 70e6a13
Author: Maximilian Michels <mxm@apache.org>
Authored: Tue Apr 19 09:20:30 2016 +0200
Committer: Maximilian Michels <mxm@apache.org>
Committed: Wed Apr 20 19:17:50 2016 +0200

----------------------------------------------------------------------
 .../flink/translation/wrappers/SourceInputFormat.java       | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dd8bc93e/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/SourceInputFormat.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/SourceInputFormat.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/SourceInputFormat.java
index dc11c77..debd1a1 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/SourceInputFormat.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/SourceInputFormat.java
@@ -46,8 +46,8 @@ public class SourceInputFormat<T> implements InputFormat<T, SourceInputSplit<T>>
   private transient PipelineOptions options;
   private final SerializedPipelineOptions serializedOptions;
 
-  private transient BoundedSource.BoundedReader<T> reader = null;
-  private boolean inputAvailable = true;
+  private transient BoundedSource.BoundedReader<T> reader;
+  private boolean inputAvailable = false;
 
   public SourceInputFormat(BoundedSource<T> initialSource, PipelineOptions options)
{
     this.initialSource = initialSource;
@@ -135,6 +135,9 @@ public class SourceInputFormat<T> implements InputFormat<T, SourceInputSplit<T>>
 
   @Override
   public void close() throws IOException {
-    reader.close();
+    // TODO null check can be removed once FLINK-3796 is fixed
+    if (reader != null) {
+      reader.close();
+    }
   }
 }


Mime
View raw message