code: public Collection<CompletableFuture<TaskManagerLocation>> getPreferredLocationsBasedOnInputs() {
  		// otherwise, base the preferred locations on the input connections
  		if (inputEdges == null) {
  			return Collections.emptySet();
  		}
  		else {
  			Set<CompletableFuture<TaskManagerLocation>> locations = new HashSet<>(getTotalNumberOfParallelSubtasks());
  			Set<CompletableFuture<TaskManagerLocation>> inputLocations = new HashSet<>(getTotalNumberOfParallelSubtasks());
  
  			// go over all inputs
  			for (int i = 0; i < inputEdges.length; i++) {
  				inputLocations.clear();
  				ExecutionEdge[] sources = inputEdges[i];
  				if (sources != null) {
  					// go over all input sources
  					for (int k = 0; k < sources.length; k++) {
  						// look-up assigned slot of input source
  						CompletableFuture<TaskManagerLocation> locationFuture = sources[k].getSource().getProducer().getCurrentTaskManagerLocationFuture();
  						// add input location
  						inputLocations.add(locationFuture);
  						// inputs which have too many distinct sources are not considered
  						if (inputLocations.size() > MAX_DISTINCT_LOCATIONS_TO_CONSIDER) {
  							inputLocations.clear();
  							break;
  						}
  					}
  				}
  				// keep the locations of the input with the least preferred locations
  				if (locations.isEmpty() || // nothing assigned yet
  						(!inputLocations.isEmpty() && inputLocations.size() < locations.size())) {
  					// current input has fewer preferred locations
  					locations.clear();
  					locations.addAll(inputLocations);
  				}
  			}
  
  			return locations.isEmpty() ? Collections.emptyList() : locations;
  		}
  	}
docs: Gets the location preferences of the vertex's current task execution, as determined by the locations
  of the predecessors from which it receives input data.
  If there are more than MAX_DISTINCT_LOCATIONS_TO_CONSIDER different locations of source data, this
  method returns {@code null} to indicate no location preference.
  
  @return The preferred locations based in input streams, or an empty iterable,
  if there is no input-based preference.