class BackPressureRequestCoordinator {

	CompletableFuture<BackPressureStats> triggerBackPressureRequest(ExecutionVertex[] tasks) {
		checkNotNull(tasks, "Tasks to request must not be null.");
		checkArgument(tasks.length >= 1, "No tasks to request.");

		// Execution IDs of running tasks
		ExecutionAttemptID[] triggerIds = new ExecutionAttemptID[tasks.length];
		Execution[] executions = new Execution[tasks.length];

		// Check that all tasks are RUNNING before triggering anything. The
		// triggering can still fail.
		for (int i = 0; i < triggerIds.length; i++) {
			Execution execution = tasks[i].getCurrentExecutionAttempt();
			if (execution != null && execution.getState() == ExecutionState.RUNNING) {
				executions[i] = execution;
				triggerIds[i] = execution.getAttemptId();
			} else {
				return FutureUtils.completedExceptionally(new IllegalStateException("Task " + tasks[i]
					.getTaskNameWithSubtaskIndex() + " is not running."));
			}
		}

		synchronized (lock) {
			if (isShutDown) {
				return FutureUtils.completedExceptionally(new IllegalStateException("Shut down."));
			}

			int requestId = requestIdCounter++;

			LOG.debug("Triggering task back pressure request {}.", requestId);

			PendingBackPressureRequest pending = new PendingBackPressureRequest(requestId, triggerIds);

			// Add the pending request before scheduling the discard task to
			// prevent races with removing it again.
			pendingRequests.put(requestId, pending);

			requestBackPressure(executions, requestId);

			return pending.getBackPressureStatsFuture();
		}
	}

	public  BackPressureRequestCoordinator(
			Executor executor,
			long requestTimeout);

}