You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2015/07/28 16:53:43 UTC
flink git commit: [FLINK-2419] [hotfix] addSink now uses transform +
remove double checkpoint commit at head operator
Repository: flink
Updated Branches:
refs/heads/master 4b44e02dd -> 78fd2146d
[FLINK-2419] [hotfix] addSink now uses transform + remove double checkpoint commit at head operator
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/78fd2146
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/78fd2146
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/78fd2146
Branch: refs/heads/master
Commit: 78fd2146dd00da1130910d9f23f09e2504854ef7
Parents: 4b44e02
Author: Gyula Fora <gy...@apache.org>
Authored: Tue Jul 28 15:47:30 2015 +0200
Committer: Gyula Fora <gy...@apache.org>
Committed: Tue Jul 28 15:50:14 2015 +0200
----------------------------------------------------------------------
.../flink/streaming/api/datastream/DataStream.java | 12 +-----------
.../flink/streaming/runtime/tasks/StreamTask.java | 14 +++++---------
2 files changed, 6 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/78fd2146/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index 7896169..5bd3fb8 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -1162,17 +1162,7 @@ public class DataStream<OUT> {
* @return The closed DataStream.
*/
public DataStreamSink<OUT> addSink(SinkFunction<OUT> sinkFunction) {
-
- OneInputStreamOperator<OUT, Object> sinkOperator = new StreamSink<OUT>(clean(sinkFunction));
-
- DataStreamSink<OUT> returnStream = new DataStreamSink<OUT>(environment, "sink", getType(),
- sinkOperator);
-
- streamGraph.addOperator(returnStream.getId(), sinkOperator, getType(), null, "Stream Sink");
-
- this.connectGraph(this.copy(), returnStream.getId(), 0);
-
- return returnStream;
+ return new DataStreamSink<OUT>((DataStream<OUT>) transform("StreamSink", null, new StreamSink<OUT>(clean(sinkFunction))));
}
private void validateUnion(Integer id) {
http://git-wip-us.apache.org/repos/asf/flink/blob/78fd2146/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 286202f..75bdd57 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -279,22 +279,18 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
}
- @SuppressWarnings({ "unchecked", "rawtypes" })
+ @SuppressWarnings("rawtypes")
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
// we do nothing here so far. this should call commit on the source function, for example
synchronized (checkpointLock) {
- if (streamOperator instanceof StatefulStreamOperator) {
- ((StatefulStreamOperator) streamOperator).notifyCheckpointComplete(checkpointId);
- }
- if (hasChainedOperators) {
- for (StreamOperator<?> chainedOperator : outputHandler.getChainedOperators()) {
- if (chainedOperator instanceof StatefulStreamOperator) {
- ((StatefulStreamOperator) chainedOperator).notifyCheckpointComplete(checkpointId);
- }
+ for (StreamOperator<?> chainedOperator : outputHandler.getChainedOperators()) {
+ if (chainedOperator instanceof StatefulStreamOperator) {
+ ((StatefulStreamOperator) chainedOperator).notifyCheckpointComplete(checkpointId);
}
}
+
}
}