code: @VisibleForTesting
  	public CompletableFuture<Collection<TaskManagerLocation>> calculatePreferredLocations(LocationPreferenceConstraint locationPreferenceConstraint) {
  		final Collection<CompletableFuture<TaskManagerLocation>> preferredLocationFutures = getVertex().getPreferredLocations();
  		final CompletableFuture<Collection<TaskManagerLocation>> preferredLocationsFuture;
  
  		switch(locationPreferenceConstraint) {
  			case ALL:
  				preferredLocationsFuture = FutureUtils.combineAll(preferredLocationFutures);
  				break;
  			case ANY:
  				final ArrayList<TaskManagerLocation> completedTaskManagerLocations = new ArrayList<>(preferredLocationFutures.size());
  
  				for (CompletableFuture<TaskManagerLocation> preferredLocationFuture : preferredLocationFutures) {
  					if (preferredLocationFuture.isDone() && !preferredLocationFuture.isCompletedExceptionally()) {
  						final TaskManagerLocation taskManagerLocation = preferredLocationFuture.getNow(null);
  
  						if (taskManagerLocation == null) {
  							throw new FlinkRuntimeException("TaskManagerLocationFuture was completed with null. This indicates a programming bug.");
  						}
  
  						completedTaskManagerLocations.add(taskManagerLocation);
  					}
  				}
  
  				preferredLocationsFuture = CompletableFuture.completedFuture(completedTaskManagerLocations);
  				break;
  			default:
  				throw new RuntimeException("Unknown LocationPreferenceConstraint " + locationPreferenceConstraint + '.');
  		}
  
  		return preferredLocationsFuture;
  	}
docs: Calculates the preferred locations based on the location preference constraint.
  
  @param locationPreferenceConstraint constraint for the location preference
  @return Future containing the collection of preferred locations. This might not be completed if not all inputs
  have been a resource assigned.