code: @Override
    public NumberedSplit<InputSplit> getInputSplit(final NodeDescriptor nodeDescriptor, final String evaluatorId) {
      synchronized (evaluatorToSplits) {
        if (evaluatorToSplits.containsKey(evaluatorId)) {
          LOG.log(Level.FINE, "Found an already allocated split, {0}", evaluatorToSplits.toString());
          return evaluatorToSplits.get(evaluatorId);
        }
      }
      // always first try to allocate based on the hostName
      final String hostName = nodeDescriptor.getName();
      LOG.log(Level.FINE, "Allocated split not found, trying on {0}", hostName);
      if (locationToSplits.containsKey(hostName)) {
        LOG.log(Level.FINE, "Found splits possibly hosted for {0} at {1}", new Object[] {evaluatorId, hostName});
        final NumberedSplit<InputSplit> split = allocateSplit(evaluatorId, locationToSplits.get(hostName));
        if (split != null) {
          return split;
        }
      }
      LOG.log(Level.FINE, "{0} does not host any splits or someone else took splits hosted here. Picking other ones",
          hostName);
      final NumberedSplit<InputSplit> split = tryAllocate(nodeDescriptor, evaluatorId);
      if (split == null) {
        throw new RuntimeException("Unable to find an input split to evaluator " + evaluatorId);
      } else {
        LOG.log(Level.FINE, evaluatorToSplits.toString());
      }
      return split;
    }
docs: Get an input split to be assigned to this evaluator.
  <p>
  Allocates one if its not already allocated
  
  @param evaluatorId
  @return a numberedSplit
  @throws RuntimeException
  if couldn't find any split