docs: @Override
  	protected void consume() {
  		int size = 0;
  		LinkedHashSet<E> batch = new LinkedHashSet<E>();
  		E obj = null;
  		try {
  			obj = queue.take();
  		} catch (InterruptedException e) {
  			return;
  		}
  		batch.add(obj);
  		size++;
  		if (pollTimeout == 0){
  			// no waiting
  			while(size < maxBatchSize && (obj = queue.poll()) != null){
  				batch.add(obj);
  				size++;
  			}
  		}else{
  			// need to wait for a while
  			try {
  				while(size < maxBatchSize && (obj = queue.poll(pollTimeout, pollTimeoutUnit)) != null){
  					batch.add(obj);
  					size++;
  				}
  			} catch (InterruptedException e) {
  				// do nothing because we need to have the batch processed;
  			}
  		}
  		process(batch);
  	}