[hotfix] [streaming] Fix race in stream tasks when canceling tasks ea…

…rly.
PangZhi · Aug 2, 2015 · 40eef52 · 40eef52
1 parent af88aa0
commit 40eef52
Show file tree

Hide file tree

Showing 5 changed files with 5 additions and 9 deletions.
diff --git a/...aming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java b/...aming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java
@@ -61,8 +61,6 @@ public void registerInputOutput() {
 
 	@Override
 	public void invoke() throws Exception {
-		this.isRunning = true;
-
 		boolean operatorOpen = false;
 
 		if (LOG.isDebugEnabled()) {

diff --git a/...reaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java b/...reaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -45,8 +45,6 @@ public class SourceStreamTask<OUT> extends StreamTask<OUT, StreamSource<OUT>> {
 	public void invoke() throws Exception {
 		final SourceOutput<StreamRecord<OUT>> output = new SourceOutput<StreamRecord<OUT>>(outputHandler.getOutput(), checkpointLock);
 
-		this.isRunning = true;
-
 		boolean operatorOpen = false;
 
 		if (LOG.isDebugEnabled()) {

diff --git a/...ming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java b/...ming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
@@ -72,7 +72,6 @@ public void registerInputOutput() {
 	@SuppressWarnings("unchecked")
 	@Override
 	public void invoke() throws Exception {
-		isRunning = true;
 		if (LOG.isDebugEnabled()) {
 			LOG.debug("Iteration source {} invoked", getName());
 		}

diff --git a/...ink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/...ink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -66,7 +66,8 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
 
 	protected boolean hasChainedOperators;
 
-	protected volatile boolean isRunning = false;
+	// needs to be initialized to true, so that early cancel() before invoke() behaves correctly
+	protected volatile boolean isRunning = true;
 
 	protected List<StreamingRuntimeContext> contexts;
 
@@ -229,10 +230,12 @@ public void setInitialState(StateHandle<Serializable> stateHandle) throws Except
 	@Override
 	public void triggerCheckpoint(long checkpointId, long timestamp) throws Exception {
 
+		LOG.debug("Starting checkpoint {} on task {}", checkpointId, getName());
+
 		synchronized (checkpointLock) {
 			if (isRunning) {
 				try {
-					LOG.debug("Starting checkpoint {} on task {}", checkpointId, getName());
+
 
 					// We wrap the states of the chained operators in a list, marking non-stateful oeprators with null
 					List<Tuple2<StateHandle<Serializable>, Map<String, OperatorStateHandle>>> chainedStates = new ArrayList<Tuple2<StateHandle<Serializable>, Map<String, OperatorStateHandle>>>();

diff --git a/...aming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java b/...aming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java
@@ -85,8 +85,6 @@ public void registerInputOutput() {
 
 	@Override
 	public void invoke() throws Exception {
-		this.isRunning = true;
-
 		boolean operatorOpen = false;
 
 		if (LOG.isDebugEnabled()) {