class RestartPipelinedRegionFailoverStrategy implements FailoverStrategy {

	@Override
	public Set<ExecutionVertexID> getTasksNeedingRestart(ExecutionVertexID executionVertexId, Throwable cause) {
		LOG.info("Calculating tasks to restart to recover the failed task {}.", executionVertexId);

		final SchedulingPipelinedRegion failedRegion = topology.getPipelinedRegionOfVertex(executionVertexId);
		if (failedRegion == null) {
			// TODO: show the task name in the log
			throw new IllegalStateException("Can not find the failover region for task " + executionVertexId, cause);
		}

		// if the failure cause is data consumption error, mark the corresponding data partition to be failed,
		// so that the failover process will try to recover it
		Optional<PartitionException> dataConsumptionException = ExceptionUtils.findThrowable(
			cause, PartitionException.class);
		if (dataConsumptionException.isPresent()) {
			resultPartitionAvailabilityChecker.markResultPartitionFailed(
				dataConsumptionException.get().getPartitionId().getPartitionId());
		}

		// calculate the tasks to restart based on the result of regions to restart
		Set<ExecutionVertexID> tasksToRestart = new HashSet<>();
		for (SchedulingPipelinedRegion region : getRegionsToRestart(failedRegion)) {
			region.getVertices().forEach(vertex -> tasksToRestart.add(vertex.getId()));
		}

		// the previous failed partition will be recovered. remove its failed state from the checker
		if (dataConsumptionException.isPresent()) {
			resultPartitionAvailabilityChecker.removeResultPartitionFromFailedState(
				dataConsumptionException.get().getPartitionId().getPartitionId());
		}

		LOG.info("{} tasks should be restarted to recover the failed task {}. ", tasksToRestart.size(), executionVertexId);
		return tasksToRestart;
	}

	@VisibleForTesting public  RestartPipelinedRegionFailoverStrategy(SchedulingTopology topology);
	public  RestartPipelinedRegionFailoverStrategy(
		SchedulingTopology topology,
		ResultPartitionAvailabilityChecker resultPartitionAvailabilityChecker);

}