docs: protected void drainToCount(final int targetCount, final boolean doShutdown, /* Boolean = false */
                      final boolean useHandoff,  /* : Boolean = config.useSoftHandoff */
                      final CountDownLatch latch)
     {
         String msg = useHandoff ? " with handoff" : "";
         LOG.info(String.format("Draining %s%s. Target count: %s, Current: %s",
                 config.workUnitName, msg, targetCount, cluster.myWorkUnits.size()));
  
         if (targetCount >= cluster.myWorkUnits.size()) {
             if (!doShutdown) {
                 return;
             }
         } else if (targetCount == 0 && doShutdown) {
             cluster.completeShutdown();
         }
  
         final int amountToDrain = cluster.myWorkUnits.size() - targetCount;
  
         String msgPrefix = (useHandoff) ? "Requesting handoff for" : "Shutting down";
         LOG.info("{} {} of {} {} over {} seconds", 
               msgPrefix, amountToDrain, cluster.myWorkUnits.size(), config.workUnitName, config.drainTime);
  
         // Build a list of work units to hand off.
         LinkedHashSet<String> wuList = new LinkedHashSet<String>(cluster.myWorkUnits);
         wuList.removeAll(cluster.workUnitsPeggedToMe);
  
         final ArrayList<String> toHandOff = new ArrayList<String>();
         Iterator<String> it = wuList.iterator();
         for (int i = amountToDrain; it.hasNext() && i > 0; --i) {
             toHandOff.add(it.next());
         }
  
         final int drainInterval = (int) (((double) config.drainTime / (double) toHandOff.size()) * 1000);
         final Iterator<String> drainIt = toHandOff.iterator();
  
         TimerTask handoffTask = new TimerTask() {
             @Override
             public void run() {
                 if (!drainIt.hasNext()) {
                     if (targetCount == 0 && doShutdown)  {
                         cluster.completeShutdown();
                     }
                     if (latch != null) {
                         latch.countDown();
                     }
                     return;
                 }
                 String workUnit = drainIt.next();
                 if (useHandoff && !isPeggedToMe(workUnit)) {
                     try {
                         cluster.requestHandoff(workUnit);
                     } catch (Exception e) {
                         LOG.warn("Problems trying to request handoff of "+workUnit, e);
                     }
                 } else {
                     cluster.shutdownWork(workUnit, true);
                 }
                 cluster.schedule(this, drainInterval, TimeUnit.MILLISECONDS);
             }
         };
  
         LOG.info("Releasing {} / {} work units over {} seconds: {}",
                 amountToDrain, cluster.myWorkUnits.size(), config.drainTime,
                 Strings.mkstring(toHandOff, ", "));
  
         if (!cluster.myWorkUnits.isEmpty()) {
             cluster.schedule(handoffTask, 0, TimeUnit.SECONDS);
         }
     }