From e8faff795005205ca877f8e9c4588e5bde42fd65 Mon Sep 17 00:00:00 2001
From: Gyula Fora <gyfora@apache.org>
Date: Tue, 7 Apr 2015 22:17:18 +0200
Subject: [PATCH] [FLINK-1813] Avoid illegalstate exception when trying to
 broadcast on finished operators

Closes #577
---
 .../api/streamvertex/StreamVertex.java        | 19 ++++++++++---------
 1 file changed, 10 insertions(+), 9 deletions(-)

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 5b805313dc744..b56eda38ad223 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 @@ protected void initialize() {
 
 	@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 void invoke() throws Exception {
 			}
 			throw e;
 		} finally {
+			this.isRunning = false;
 			// Cleanup
 			outputHandler.flushOutputs();
 			clearBuffers();
-			this.isRunning = false;
 		}
 
 	}
@@ -288,7 +286,7 @@ public EventListener<TaskEvent> getSuperstepListener() {
 	 * @param id
 	 */
 	private synchronized void actOnBarrier(long id) {
-		if (this.isRunning) {
+		if (isRunning) {
 			try {
 				outputHandler.broadcastBarrier(id);
 				confirmBarrier(id);
@@ -296,7 +294,10 @@ private synchronized void actOnBarrier(long id) {
 					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);
+				}
 			}
 		}
 	}