[streaming] Temporal fix for streaming source parallelism
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/2467f36c Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/2467f36c Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/2467f36c Branch: refs/heads/release-0.8 Commit: 2467f36c80830e83b43271c89cf1ec827882b424 Parents: 6b3c3a1 Author: mbalassi <[email protected]> Authored: Fri Dec 26 18:06:51 2014 +0100 Committer: mbalassi <[email protected]> Committed: Fri Dec 26 18:06:51 2014 +0100 ---------------------------------------------------------------------- .../streaming/api/environment/StreamExecutionEnvironment.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/2467f36c/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java ---------------------------------------------------------------------- diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java index 783fa28..59d56aa 100644 --- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java +++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java @@ -379,8 +379,9 @@ public abstract class StreamExecutionEnvironment { DataStreamSource<OUT> returnStream = new DataStreamSource<OUT>(this, "source", outTypeInfo); try { + //currently parallel sources are not directly supported jobGraphBuilder.addSourceVertex(returnStream.getId(), function, null, outTypeInfo, - "source", SerializationUtils.serialize(function), getDegreeOfParallelism()); + "source", SerializationUtils.serialize(function), 1); } catch (SerializationException e) { throw new RuntimeException("Cannot serialize SourceFunction"); }
