code: @Override
  	public void onTaskFailure(Execution taskExecution, Throwable cause) {
  		final ExecutionVertex ev = taskExecution.getVertex();
  		final FailoverRegion failoverRegion = vertexToRegion.get(ev);
  
  		if (failoverRegion == null) {
  			executionGraph.failGlobal(new FlinkException(
  					"Can not find a failover region for the execution " + ev.getTaskNameWithSubtaskIndex(), cause));
  		}
  		else {
  			LOG.info("Recovering task failure for {} #{} ({}) via restart of failover region",
  					taskExecution.getVertex().getTaskNameWithSubtaskIndex(),
  					taskExecution.getAttemptNumber(),
  					taskExecution.getAttemptId());
  
  			failoverRegion.onExecutionFail(taskExecution, cause);
  		}
  	}
docs: ------------------------------------------------------------------------