Repository: flink Updated Branches: refs/heads/master 9b63f269e -> e01057e8e
[streaming] Fix bug with operators inheriting lower parallelism Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e01057e8 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e01057e8 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e01057e8 Branch: refs/heads/master Commit: e01057e8e8ebb17259afc5529dcc620bb48c21e5 Parents: 9b63f26 Author: Gyula Fora <[email protected]> Authored: Mon Jan 26 09:45:15 2015 +0100 Committer: Gyula Fora <[email protected]> Committed: Mon Jan 26 09:45:15 2015 +0100 ---------------------------------------------------------------------- .../org/apache/flink/streaming/api/datastream/DataStream.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/e01057e8/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java ---------------------------------------------------------------------- diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java index b30d261..b7ad97e 100644 --- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java +++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java @@ -1128,7 +1128,7 @@ public class DataStream<OUT> { operatorName, outTypeInfo, invokable); streamGraph.addStreamVertex(returnStream.getId(), invokable, getType(), outTypeInfo, - operatorName, degreeOfParallelism); + operatorName, returnStream.getParallelism()); connectGraph(inputStream, returnStream.getId(), 0); @@ -1195,7 +1195,7 @@ public class DataStream<OUT> { sinkInvokable); streamGraph.addStreamVertex(returnStream.getId(), sinkInvokable, getType(), null, "sink", - degreeOfParallelism); + returnStream.getParallelism()); this.connectGraph(this.copy(), returnStream.getId(), 0);
