flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From srich...@apache.org
Subject flink git commit: [FLINK-7063] [checkpoint] Call super.cancel(...) before closing streams in AsyncStoppableTaskWithCallback
Date Mon, 10 Jul 2017 14:40:54 GMT
Repository: flink
Updated Branches:
  refs/heads/master ccf10002b -> 74adb8474


[FLINK-7063] [checkpoint] Call super.cancel(...) before closing streams in AsyncStoppableTaskWithCallback


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/74adb847
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/74adb847
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/74adb847

Branch: refs/heads/master
Commit: 74adb84742f803d2d3585903ccbdde9f52ad523a
Parents: ccf1000
Author: Stefan Richter <s.richter@data-artisans.com>
Authored: Mon Jul 3 11:46:16 2017 +0200
Committer: Stefan Richter <s.richter@data-artisans.com>
Committed: Mon Jul 10 16:40:16 2017 +0200

----------------------------------------------------------------------
 .../io/async/AsyncStoppableTaskWithCallback.java      | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/74adb847/flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AsyncStoppableTaskWithCallback.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AsyncStoppableTaskWithCallback.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AsyncStoppableTaskWithCallback.java
index 1ca109c..a30c607 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AsyncStoppableTaskWithCallback.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AsyncStoppableTaskWithCallback.java
@@ -36,13 +36,21 @@ public class AsyncStoppableTaskWithCallback<V> extends FutureTask<V>
{
 
 	@Override
 	public boolean cancel(boolean mayInterruptIfRunning) {
-		stoppableCallbackCallable.stop();
-		return super.cancel(mayInterruptIfRunning);
+		final boolean cancel = super.cancel(mayInterruptIfRunning);
+		if (cancel) {
+			stoppableCallbackCallable.stop();
+			// this is where we report done() for the cancel case, after calling stop().
+			stoppableCallbackCallable.done(true);
+		}
+		return cancel;
 	}
 
 	@Override
 	protected void done() {
-		stoppableCallbackCallable.done(isCancelled());
+		// we suppress forwarding if we have not been canceled, because the cancel case will call
to this method separately.
+		if (!isCancelled()) {
+			stoppableCallbackCallable.done(false);
+		}
 	}
 
 	public static <V> AsyncStoppableTaskWithCallback<V> from(StoppableCallbackCallable<V>
callable) {


Mime
View raw message