docs: public Iterable<V> sorted(Iterator<V> input) {
          ExecutorService executor = new ThreadPoolExecutor(this.numThreads,
                                                            this.numThreads,
                                                            1000L,
                                                            TimeUnit.MILLISECONDS,
                                                            new SynchronousQueue<Runnable>(),
                                                            new CallerRunsPolicy());
          final AtomicInteger count = new AtomicInteger(0);
          final List<File> tempFiles = Collections.synchronizedList(new ArrayList<File>());
          while(input.hasNext()) {
              final int segmentId = count.getAndIncrement();
              final long segmentStartMs = System.currentTimeMillis();
              logger.info("Segment " + segmentId + ": filling sort buffer for segment...");
              @SuppressWarnings("unchecked")
              final V[] buffer = (V[]) new Object[internalSortSize];
              int segmentSizeIter = 0;
              for(; segmentSizeIter < internalSortSize && input.hasNext(); segmentSizeIter++)
                  buffer[segmentSizeIter] = input.next();
              final int segmentSize = segmentSizeIter;
              logger.info("Segment " + segmentId + ": sort buffer filled...adding to sort queue.");
  
              // sort and write out asynchronously
              executor.execute(new Runnable() {
  
                  public void run() {
                      logger.info("Segment " + segmentId + ": sorting buffer.");
                      long start = System.currentTimeMillis();
                      Arrays.sort(buffer, 0, segmentSize, comparator);
                      long elapsed = System.currentTimeMillis() - start;
                      logger.info("Segment " + segmentId + ": sort completed in " + elapsed
                                  + " ms, writing to temp file.");
  
                      // write out values to a temp file
                      try {
                          File tempFile = File.createTempFile("segment-", ".dat", tempDir);
                          tempFile.deleteOnExit();
                          tempFiles.add(tempFile);
                          OutputStream os = new BufferedOutputStream(new FileOutputStream(tempFile),
                                                                     bufferSize);
                          if(gzip)
                              os = new GZIPOutputStream(os);
                          DataOutputStream output = new DataOutputStream(os);
                          for(int i = 0; i < segmentSize; i++)
                              writeValue(output, buffer[i]);
                          output.close();
                      } catch(IOException e) {
                          throw new VoldemortException(e);
                      }
                      long segmentElapsed = System.currentTimeMillis() - segmentStartMs;
                      logger.info("Segment " + segmentId + ": completed processing of segment in "
                                  + segmentElapsed + " ms.");
                  }
              });
          }
  
          // wait for all sorting to complete
          executor.shutdown();
          try {
              executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
              // create iterator over sorted values
              return new DefaultIterable<V>(new ExternalSorterIterator(tempFiles, bufferSize
                                                                                  / tempFiles.size()));
          } catch(InterruptedException e) {
              throw new RuntimeException(e);
          }
      }