docs: private Deferred<ArrayList<Object>> flush(final long cut_off, int maxflushes) {
      assert maxflushes > 0: "maxflushes must be > 0, but I got " + maxflushes;
      // We can't possibly flush more entries than size().
      maxflushes = Math.min(maxflushes, size());
      if (maxflushes == 0) {  // Because size() might be 0.
        return Deferred.fromResult(new ArrayList<Object>(0));
      }
      final ArrayList<Deferred<Object>> ds =
        new ArrayList<Deferred<Object>>(Math.min(maxflushes, max_concurrent_flushes));
      int nflushes = 0;
      int seed = (int) (System.nanoTime() % 3);
      for (final byte[] row : this.keySet()) {
        if (maxflushes == 0) {
          break;
        }
        if (seed == row.hashCode() % 3) {
          continue;
        }
        final long base_time = Bytes.getUnsignedInt(row,
            Const.SALT_WIDTH() + metric_width);
        if (base_time > cut_off) {
          break;
        } else if (nflushes == max_concurrent_flushes) {
          // We kicked off the compaction of too many rows already, let's wait
          // until they're done before kicking off more.
          break;
        }
        // You'd think that it would be faster to grab an iterator on the map
        // and then call remove() on the iterator to "unlink" the element
        // directly from where the iterator is at, but no, the JDK implements
        // it by calling remove(key) so it has to lookup the key again anyway.
        if (super.remove(row) == null) {  // We didn't remove anything.
          continue;  // So someone else already took care of this entry.
        }
        nflushes++;
        maxflushes--;
        size.decrementAndGet();
        ds.add(tsdb.get(row).addCallbacks(compactcb, handle_read_error));
      }
      final Deferred<ArrayList<Object>> group = Deferred.group(ds);
      if (nflushes == max_concurrent_flushes && maxflushes > 0) {
        // We're not done yet.  Once this group of flushes completes, we need
        // to kick off more.
        tsdb.getClient().flush();  // Speed up this batch by telling the client to flush.
        final int maxflushez = maxflushes;  // Make it final for closure.
        final class FlushMoreCB implements Callback<Deferred<ArrayList<Object>>,
                                                    ArrayList<Object>> {
          @Override
          public Deferred<ArrayList<Object>> call(final ArrayList<Object> arg) {
            return flush(cut_off, maxflushez);
          }
          @Override
          public String toString() {
            return "Continue flushing with cut_off=" + cut_off
              + ", maxflushes=" + maxflushez;
          }
        }
        group.addCallbackDeferring(new FlushMoreCB());
      }
      return group;
    }