@Override protected BlockingBuffer<KVO<O>[]> newSubtaskBuffer(){
  return new BlockingBuffer<KVO<O>[]>(new ArrayBlockingQueue<KVO<O>[]>(subtaskQueueCapacity),BlockingBuffer.DEFAULT_MINIMUM_CHUNK_SIZE,Long.MAX_VALUE,TimeUnit.SECONDS,true);
}

@ Override protected BlockingBuffer < KVO < O > [ ] > newSubtaskBuffer ( ) { return new BlockingBuffer < KVO < O > [ ] > ( new ArrayBlockingQueue < KVO < O > [ ] > ( subtaskQueueCapacity ) , BlockingBuffer . DEFAULT_MINIMUM_CHUNK_SIZE , Long . MAX_VALUE , TimeUnit . SECONDS , true ) ; }
the subtask will use a buffer with an infinite chunk timeout .
