You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mb...@apache.org on 2014/12/26 18:39:02 UTC
[2/2] incubator-flink git commit: [streaming] Temporal fix for
streaming source parallelism
[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 <mb...@apache.org>
Authored: Fri Dec 26 18:06:51 2014 +0100
Committer: mbalassi <mb...@apache.org>
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");
}