private void awaitAll() throws InterruptedException, ExecutionException {
  if (buffer.isOpen()) {
    throw new IllegalStateException();
  }
  willShutdown();
  while (true) {
    halted();
    final E[] a;
    lock.lockInterruptibly();
    try {
      a=redirectQueue.poll();
      if (a == null) {
        if (finishedSubtaskQueue.isEmpty() && sinks.isEmpty() && redirectQueue.isEmpty()&& nothingPending()) {
          if (log.isInfoEnabled())           log.info("All subtasks are done: " + this);
          return;
        }
        if (log.isDebugEnabled())         log.debug("Waiting for " + sinks.size() + " subtasks : "+ this);
        drainFutures();
        if (!finishedSubtaskQueue.isEmpty()) {
          subtaskDone.await(50,TimeUnit.MILLISECONDS);
        }
      }
    }
  finally {
      lock.unlock();
    }
    if (a != null) {
      handleChunk(a,true);
    }
  }
}

private void awaitAll ( ) throws InterruptedException , ExecutionException { if ( buffer . isOpen ( ) ) { throw new IllegalStateException ( ) ; } willShutdown ( ) ; while ( true ) { halted ( ) ; final E [ ] a ; lock . lockInterruptibly ( ) ; try { a = redirectQueue . poll ( ) ; if ( a == null ) { if ( finishedSubtaskQueue . isEmpty ( ) && sinks . isEmpty ( ) && redirectQueue . isEmpty ( ) && nothingPending ( ) ) { if ( log . isInfoEnabled ( ) ) log . info ( "All subtasks are done: " + this ) ; return ; } if ( log . isDebugEnabled ( ) ) log . debug ( "Waiting for " + sinks . size ( ) + " subtasks : " + this ) ; drainFutures ( ) ; if ( ! finishedSubtaskQueue . isEmpty ( ) ) { subtaskDone . await ( 50 , TimeUnit . MILLISECONDS ) ; } } } finally { lock . unlock ( ) ; } if ( a != null ) { handleChunk ( a , true ) ; } } }
await the completion of the writes on each index partition . the master will terminate when there are no active subtasks and the redirect queue is empty . that condition is tested atomically .
