class ExecutionFailureHandler {

	public FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause) {
		if (isUnrecoverableError(cause)) {
			return FailureHandlingResult.unrecoverable(new JobException("The failure is not recoverable", cause));
		}

		restartBackoffTimeStrategy.notifyFailure(cause);
		if (restartBackoffTimeStrategy.canRestart()) {
			return FailureHandlingResult.restartable(
				failoverStrategy.getTasksNeedingRestart(failedTask, cause),
				restartBackoffTimeStrategy.getBackoffTime());
		} else {
			return FailureHandlingResult.unrecoverable(
				new JobException("Failed task restarting is suppressed by " + restartBackoffTimeStrategy, cause));
		}
	}

	public  ExecutionFailureHandler(
		FailoverStrategy failoverStrategy,
		RestartBackoffTimeStrategy restartBackoffTimeStrategy);

}