Fix condition in FlinkStreamingPipelineTranslator
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/1524494f Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/1524494f Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/1524494f Branch: refs/heads/gearpump-runner Commit: 1524494f77ca8667d57b00651b39005524c71d64 Parents: 6ae4b6a Author: Aljoscha Krettek <aljoscha.kret...@gmail.com> Authored: Wed Aug 31 13:42:30 2016 +0200 Committer: Dan Halperin <dhalp...@google.com> Committed: Mon Sep 12 17:40:11 2016 -0700 ---------------------------------------------------------------------- .../flink/translation/FlinkStreamingPipelineTranslator.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/1524494f/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingPipelineTranslator.java ---------------------------------------------------------------------- diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingPipelineTranslator.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingPipelineTranslator.java index b127455..284cd23 100644 --- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingPipelineTranslator.java +++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingPipelineTranslator.java @@ -84,7 +84,7 @@ public class FlinkStreamingPipelineTranslator extends FlinkPipelineTranslator { StreamTransformTranslator<?> translator = FlinkStreamingTransformTranslators.getTranslator(transform); - if (translator == null && applyCanTranslate(transform, node, translator)) { + if (translator == null || !applyCanTranslate(transform, node, translator)) { LOG.info(node.getTransform().getClass().toString()); throw new UnsupportedOperationException( "The transform " + transform + " is currently not supported.");
