This closes apache#3263

mransley · May 31, 2017 · 4884d48 · 4884d48
2 parents 19c33df + 5780fc5
commit 4884d48
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/.../src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java b/.../src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
@@ -90,7 +90,7 @@ public void mapPartition(
  RuntimeContext runtimeContext = getRuntimeContext();
 
  DoFnRunners.OutputManager outputManager;
- if (outputMap == null) {
+ if (outputMap.size() == 1) {
  outputManager = new FlinkDoFnFunction.DoFnOutputManager(out);
  } else {
  // it has some additional outputs
@@ -146,7 +146,9 @@ static class DoFnOutputManager
  @Override
  @SuppressWarnings("unchecked")
  public <T> void output(TupleTag<T> tag, WindowedValue<T> output) {
- collector.collect(output);
+ collector.collect(
+ WindowedValue.of(new RawUnionValue(0 /* single output */, output.getValue()),
+ output.getTimestamp(), output.getWindows(), output.getPane()));
  }
  }
 

diff --git a/...n/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java b/...n/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java
@@ -91,7 +91,7 @@ public void reduce(
  RuntimeContext runtimeContext = getRuntimeContext();
 
  DoFnRunners.OutputManager outputManager;
- if (outputMap == null) {
+ if (outputMap.size() == 1) {
  outputManager = new FlinkDoFnFunction.DoFnOutputManager(out);
  } else {
  // it has some additional Outputs