protected boolean handleChunk(final E[] chunk) throws ExecutionException, InterruptedException, IOException {
  assert chunk != null;
  assert chunk.length > 0;
  final int chunkSize=chunk.length;
  final long beginNanos=System.nanoTime();
  try {
    for (    E e : chunk) {
      master.addPending(e,this,locator);
    }
    try {
      clientTask.accept(chunk);
    }
 catch (    Throwable t) {
      master.halt(t);
      throw new RuntimeException(t);
    }
    if (log.isDebugEnabled())     log.debug(stats);
    return false;
  }
  finally {
    final long elapsedNanos=System.nanoTime() - beginNanos;
synchronized (stats) {
      stats.chunksOut.incrementAndGet();
      stats.elementsOut.addAndGet(chunkSize);
      stats.elapsedChunkWritingNanos+=elapsedNanos;
    }
synchronized (master.stats) {
      master.stats.chunksOut.incrementAndGet();
      master.stats.elementsOut.addAndGet(chunkSize);
      master.stats.elapsedSinkChunkWritingNanos+=elapsedNanos;
    }
  }
}

protected boolean handleChunk ( final E [ ] chunk ) throws ExecutionException , InterruptedException , IOException { assert chunk != null ; assert chunk . length > 0 ; final int chunkSize = chunk . length ; final long beginNanos = System . nanoTime ( ) ; try { for ( E e : chunk ) { master . addPending ( e , this , locator ) ; } try { clientTask . accept ( chunk ) ; } catch ( Throwable t ) { master . halt ( t ) ; throw new RuntimeException ( t ) ; } if ( log . isDebugEnabled ( ) ) log . debug ( stats ) ; return false ; } finally { final long elapsedNanos = System . nanoTime ( ) - beginNanos ; synchronized ( stats ) { stats . chunksOut . incrementAndGet ( ) ; stats . elementsOut . addAndGet ( chunkSize ) ; stats . elapsedChunkWritingNanos += elapsedNanos ; } synchronized ( master . stats ) { master . stats . chunksOut . incrementAndGet ( ) ; master . stats . elementsOut . addAndGet ( chunkSize ) ; master . stats . elapsedSinkChunkWritingNanos += elapsedNanos ; } } }
submits the chunk of resources for processing by the remote client task . clients should accept resources for asynchronous processing , notifying the sink as resources succeed or fail .
