Repository: beam Updated Branches: refs/heads/master 19c33dfa6 -> 4884d4867
Flink*DoFnFunction: fix check for single-output dofns Fixes Findbugs and (presumably) increases efficiency Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/a0444b8c Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/a0444b8c Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/a0444b8c Branch: refs/heads/master Commit: a0444b8ce7f6d39b039612190102e146ef4148dd Parents: 19c33df Author: Dan Halperin <dhalp...@google.com> Authored: Tue May 30 16:12:23 2017 -0700 Committer: Dan Halperin <dhalp...@google.com> Committed: Wed May 31 11:36:15 2017 -0700 ---------------------------------------------------------------------- .../runners/flink/translation/functions/FlinkDoFnFunction.java | 2 +- .../flink/translation/functions/FlinkStatefulDoFnFunction.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/a0444b8c/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java ---------------------------------------------------------------------- diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java index 42a8833..ab2ac6b 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java @@ -90,7 +90,7 @@ public class FlinkDoFnFunction<InputT, OutputT> RuntimeContext runtimeContext = getRuntimeContext(); DoFnRunners.OutputManager outputManager; - if (outputMap == null) { + if (outputMap.size() == 1) { outputManager = new FlinkDoFnFunction.DoFnOutputManager(out); } else { // it has some additional outputs http://git-wip-us.apache.org/repos/asf/beam/blob/a0444b8c/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java ---------------------------------------------------------------------- diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java index b075768..11d4fee 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java @@ -91,7 +91,7 @@ public class FlinkStatefulDoFnFunction<K, V, OutputT> RuntimeContext runtimeContext = getRuntimeContext(); DoFnRunners.OutputManager outputManager; - if (outputMap == null) { + if (outputMap.size() == 1) { outputManager = new FlinkDoFnFunction.DoFnOutputManager(out); } else { // it has some additional Outputs
