Flink Schedule Mode

1
/** Schedule tasks lazy from the sources. Downstream tasks start once their input data are ready */
2
LAZY_FROM_SOURCES,
3

4
/** Schedules all tasks immediately. */
5
EAGER;

1
private CompletableFuture<Void> scheduleLazy(SlotProvider slotProvider) {
2

3
  final ArrayList<CompletableFuture<Void>> schedulingFutures = new ArrayList<>(numVerticesTotal);
4

5
  // take the vertices without inputs.
6
  for (ExecutionJobVertex ejv : verticesInCreationOrder) {
7
    if (ejv.getJobVertex().isInputVertex()) {
8
      final CompletableFuture<Void> schedulingJobVertexFuture = ejv.scheduleAll(
9
        slotProvider,
10
        allowQueuedScheduling,
11
        LocationPreferenceConstraint.ALL); // since it is an input vertex, the input based location preferences should be empty
12

13
      schedulingFutures.add(schedulingJobVertexFuture);
14
    }
15
  }
16

17
  return FutureUtils.waitForAll(schedulingFutures);
18
}

1
private CompletableFuture<Void> scheduleEager(SlotProvider slotProvider, final Time timeout) {
2
  checkState(state == JobStatus.RUNNING, "job is not running currently");
3

4
  // Important: reserve all the space we need up front.
5
  // that way we do not have any operation that can fail between allocating the slots
6
  // and adding them to the list. If we had a failure in between there, that would
7
  // cause the slots to get lost
8
  final boolean queued = allowQueuedScheduling;
9

10
  // collecting all the slots may resize and fail in that operation without slots getting lost
11
  final ArrayList<CompletableFuture<Execution>> allAllocationFutures = new ArrayList<>(getNumberOfExecutionJobVertices());
12

13
  // allocate the slots (obtain all their futures
14
  for (ExecutionJobVertex ejv : getVerticesTopologically()) {
15
    // these calls are not blocking, they only return futures
16
    Collection<CompletableFuture<Execution>> allocationFutures = ejv.allocateResourcesForAll(
17
      slotProvider,
18
      queued,
19
      LocationPreferenceConstraint.ALL,
20
      allocationTimeout);
21

22
    allAllocationFutures.addAll(allocationFutures);
23
  }
24

25
  // this future is complete once all slot futures are complete.
26
  // the future fails once one slot future fails.
27
  final ConjunctFuture<Collection<Execution>> allAllocationsFuture = FutureUtils.combineAll(allAllocationFutures);
28

29
  final CompletableFuture<Void> currentSchedulingFuture = allAllocationsFuture
30
    .thenAccept(
31
      (Collection<Execution> executionsToDeploy) -> {
32
        for (Execution execution : executionsToDeploy) {
33
          try {
34
            execution.deploy();
35
          } catch (Throwable t) {
36
            throw new CompletionException(
37
              new FlinkException(
38
                String.format("Could not deploy execution %s.", execution),
39
                t));
40
          }
41
        }
42
      })
43
    // Generate a more specific failure message for the eager scheduling
44
    .exceptionally(
45
      (Throwable throwable) -> {
46
        final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable);
47
        final Throwable resultThrowable;
48

49
        if (strippedThrowable instanceof TimeoutException) {
50
          int numTotal = allAllocationsFuture.getNumFuturesTotal();
51
          int numComplete = allAllocationsFuture.getNumFuturesCompleted();
52
          String message = "Could not allocate all requires slots within timeout of " +
53
            timeout + ". Slots required: " + numTotal + ", slots allocated: " + numComplete;
54

55
          resultThrowable = new NoResourceAvailableException(message);
56
        } else {
57
          resultThrowable = strippedThrowable;
58
        }
59

60
        throw new CompletionException(resultThrowable);
61
      });
62

63
  return currentSchedulingFuture;
64
}