docs: @Override
  	public LocatableInputSplit getNextInputSplit(String host, int taskId) {
  
  		// for a null host, we return a remote split
  		if (host == null) {
  			synchronized (this.remoteSplitChooser) {
  				synchronized (this.unassigned) {
  
  					LocatableInputSplitWithCount split = this.remoteSplitChooser.getNextUnassignedMinLocalCountSplit(this.unassigned);
  
  					if (split != null) {
  						// got a split to assign. Double check that it hasn't been assigned before.
  						if (this.unassigned.remove(split)) {
  							if (LOG.isInfoEnabled()) {
  								LOG.info("Assigning split to null host (random assignment).");
  							}
  
  							remoteAssignments++;
  							return split.getSplit();
  						} else {
  							throw new IllegalStateException("Chosen InputSplit has already been assigned. This should not happen!");
  						}
  					} else {
  						// all splits consumed
  						if (LOG.isDebugEnabled()) {
  							LOG.debug("No more unassigned input splits remaining.");
  						}
  						return null;
  					}
  				}
  			}
  		}
  
  		host = host.toLowerCase(Locale.US);
  
  		// for any non-null host, we take the list of non-null splits
  		LocatableInputSplitChooser localSplits = this.localPerHost.get(host);
  
  		// if we have no list for this host yet, create one
  		if (localSplits == null) {
  			localSplits = new LocatableInputSplitChooser();
  
  			// lock the list, to be sure that others have to wait for that host's local list
  			synchronized (localSplits) {
  				LocatableInputSplitChooser prior = this.localPerHost.putIfAbsent(host, localSplits);
  
  				// if someone else beat us in the case to create this list, then we do not populate this one, but
  				// simply work with that other list
  				if (prior == null) {
  					// we are the first, we populate
  
  					// first, copy the remaining splits to release the lock on the set early
  					// because that is shared among threads
  					LocatableInputSplitWithCount[] remaining;
  					synchronized (this.unassigned) {
  						remaining = this.unassigned.toArray(new LocatableInputSplitWithCount[this.unassigned.size()]);
  					}
  
  					for (LocatableInputSplitWithCount isw : remaining) {
  						if (isLocal(host, isw.getSplit().getHostnames())) {
  							// Split is local on host.
  							// Increment local count
  							isw.incrementLocalCount();
                              // and add to local split list
  							localSplits.addInputSplit(isw);
  						}
  					}
  
  				}
  				else {
  					// someone else was faster
  					localSplits = prior;
  				}
  			}
  		}
  
  
  		// at this point, we have a list of local splits (possibly empty)
  		// we need to make sure no one else operates in the current list (that protects against
  		// list creation races) and that the unassigned set is consistent
  		// NOTE: we need to obtain the locks in this order, strictly!!!
  		synchronized (localSplits) {
  			synchronized (this.unassigned) {
  
  				LocatableInputSplitWithCount split = localSplits.getNextUnassignedMinLocalCountSplit(this.unassigned);
  
  				if (split != null) {
  					// found a valid split. Double check that it hasn't been assigned before.
  					if (this.unassigned.remove(split)) {
  						if (LOG.isInfoEnabled()) {
  							LOG.info("Assigning local split to host " + host);
  						}
  
  						localAssignments++;
  						return split.getSplit();
  					} else {
  						throw new IllegalStateException("Chosen InputSplit has already been assigned. This should not happen!");
  					}
  				}
  			}
  		}
  
  		// we did not find a local split, return a remote split
  		synchronized (this.remoteSplitChooser) {
  			synchronized (this.unassigned) {
  				LocatableInputSplitWithCount split = this.remoteSplitChooser.getNextUnassignedMinLocalCountSplit(this.unassigned);
  
  				if (split != null) {
  					// found a valid split. Double check that it hasn't been assigned yet.
  					if (this.unassigned.remove(split)) {
  						if (LOG.isInfoEnabled()) {
  							LOG.info("Assigning remote split to host " + host);
  						}
  
  						remoteAssignments++;
  						return split.getSplit();
  					} else {
  						throw new IllegalStateException("Chosen InputSplit has already been assigned. This should not happen!");
  					}
  				} else {
  					// all splits consumed
  					if (LOG.isDebugEnabled()) {
  						LOG.debug("No more input splits remaining.");
  					}
  					return null;
  				}
  			}
  		}
  	}