[FLINK-7063] [checkpoint] Call super.cancel(...) before closing strea…

…ms in AsyncStoppableTaskWithCallback

flink-runtime/src/main/java/org/apache/flink/runtime/io/async/AsyncStoppableTaskWithCallback.java

@@ -36,13 +36,21 @@ public AsyncStoppableTaskWithCallback(StoppableCallbackCallable<V> callable) {
 
 	@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) {