[Distributed runtime] Fail task if execution state update is unsuccessful

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java

@@ -729,6 +729,14 @@ public class TaskManager implements TaskOperationProtocol {
 		boolean success = false;
 		try {
 			success = this.jobManager.updateTaskExecutionState(new TaskExecutionState(jobID, executionId, newExecutionState, optionalError));
+
+			if (!success) {
+				Task task = runningTasks.get(executionId);
+
+				if (task != null) {
+					task.failExternally(new IllegalStateException("Task has been disposed on JobManager."));
+				}
+			}
 		}
 		catch (Throwable t) {
 			String msg = "Error sending task state update to JobManager.";