code: protected NumberedSplit<InputSplit> allocateSplit(final String evaluatorId,
        final BlockingQueue<NumberedSplit<InputSplit>> value) {
      if (value == null) {
        LOG.log(Level.FINE, "Queue of splits can't be empty. Returning null");
        return null;
      }
      while (true) {
        final NumberedSplit<InputSplit> split = value.poll();
        if (split == null) {
          return null;
        }
        if (value == unallocatedSplits || unallocatedSplits.remove(split)) {
          LOG.log(Level.FINE, "Found split-" + split.getIndex() + " in the queue");
          final NumberedSplit<InputSplit> old = evaluatorToSplits.putIfAbsent(evaluatorId, split);
          if (old != null) {
            throw new RuntimeException("Trying to assign different splits to the same evaluator is not supported");
          } else {
            LOG.log(Level.FINE, "Returning " + split.getIndex());
            return split;
          }
        }
      }
    }
docs: Allocates the first available split into the evaluator.
  
  @param evaluatorId
  the evaluator id
  @param value
  the queue of splits
  @return a numberedSplit or null if it cannot find one