[FLINK-15317][runtime] Handle closed input gates more gracefully duri…

…ng cancellation. (apache#10621)

During task cancellation, input gates can be closed in the task cancellation thread, before the main invokable has been interrupted, which resulted in reads from a closed gate.

This commit will handle this case in a more graceful manner such that the end-user is not seeing any additional exception.

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java

@@ -22,6 +22,7 @@
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.event.AbstractEvent;
 import org.apache.flink.runtime.event.TaskEvent;
+import org.apache.flink.runtime.execution.CancelTaskException;
 import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
 import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
 import org.apache.flink.runtime.io.network.buffer.Buffer;
@@ -480,7 +481,7 @@ private Optional<BufferOrEvent> getNextBufferOrEvent(boolean blocking) throws IO
 		}
 
 		if (closeFuture.isDone()) {
-			throw new IllegalStateException("Released");
+			throw new CancelTaskException("Input gate is already closed.");
 		}
 
 		Optional<InputWithData<InputChannel, BufferAndAvailability>> next = waitAndGetNextData(blocking);

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java

@@ -503,6 +503,11 @@ private void runMailboxLoop() throws Exception {
 				throw e;
 			}
 		}
+		catch (CancelTaskException e) {
+			if (!canceled) {
+				LOG.error("Received CancelTaskException while we are not canceled. This is a bug and should be reported", e);
+			}
+		}
 		catch (Exception e) {
 			if (canceled) {
 				LOG.warn("Error while canceling task.", e);