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/04/08 15:25:39 UTC
flink git commit: [FLINK-1813] Avoid illegalstate exception when
trying to broadcast on finished operators
Repository: flink
Updated Branches:
refs/heads/master 41ab43604 -> e8faff795
[FLINK-1813] Avoid illegalstate exception when trying to broadcast on finished operators
Closes #577
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e8faff79
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e8faff79
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e8faff79
Branch: refs/heads/master
Commit: e8faff795005205ca877f8e9c4588e5bde42fd65
Parents: 41ab436
Author: Gyula Fora <gy...@apache.org>
Authored: Tue Apr 7 22:17:18 2015 +0200
Committer: Gyula Fora <gy...@apache.org>
Committed: Wed Apr 8 15:23:56 2015 +0200
----------------------------------------------------------------------
.../streaming/api/streamvertex/StreamVertex.java | 19 ++++++++++---------
1 file changed, 10 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e8faff79/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java
index 5b80531..b56eda3 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java
@@ -96,13 +96,11 @@ public class StreamVertex<IN, OUT> extends AbstractInvokable implements StreamTa
@Override
public void broadcastBarrierFromSource(long id) {
- if (this.isRunning) {
- // Only called at input vertices
- if (LOG.isDebugEnabled()) {
- LOG.debug("Received barrier from jobmanager: " + id);
- }
- actOnBarrier(id);
+ // Only called at input vertices
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Received barrier from jobmanager: " + id);
}
+ actOnBarrier(id);
}
/**
@@ -190,10 +188,10 @@ public class StreamVertex<IN, OUT> extends AbstractInvokable implements StreamTa
}
throw e;
} finally {
+ this.isRunning = false;
// Cleanup
outputHandler.flushOutputs();
clearBuffers();
- this.isRunning = false;
}
}
@@ -288,7 +286,7 @@ public class StreamVertex<IN, OUT> extends AbstractInvokable implements StreamTa
* @param id
*/
private synchronized void actOnBarrier(long id) {
- if (this.isRunning) {
+ if (isRunning) {
try {
outputHandler.broadcastBarrier(id);
confirmBarrier(id);
@@ -296,7 +294,10 @@ public class StreamVertex<IN, OUT> extends AbstractInvokable implements StreamTa
LOG.debug("Superstep " + id + " processed: " + StreamVertex.this);
}
} catch (Exception e) {
- throw new RuntimeException(e);
+ // Only throw any exception if the vertex is still running
+ if (isRunning) {
+ throw new RuntimeException(e);
+ }
}
}
}