code: public Batch<D> getNextAvailableBatch () {
      final ReentrantLock lock = SequentialBasedBatchAccumulator.this.dqLock;
      try {
        lock.lock();
        if (SequentialBasedBatchAccumulator.this.isClosed()) {
          return dq.poll();
        } else {
            while (dq.size() == 0) {
              LOG.debug ("ready to sleep because of queue is empty");
              SequentialBasedBatchAccumulator.this.notEmpty.await();
              if (SequentialBasedBatchAccumulator.this.isClosed()) {
                return dq.poll();
              }
            }
  
            if (dq.size() > 1) {
              BytesBoundedBatch candidate = dq.poll();
              SequentialBasedBatchAccumulator.this.notFull.signal();
              LOG.debug ("retrieve batch " + candidate.getId());
              return candidate;
            }
  
            if (dq.size() == 1) {
              if (dq.peekFirst().isTTLExpire()) {
                LOG.debug ("Batch " + dq.peekFirst().getId() + " is expired");
                BytesBoundedBatch candidate = dq.poll();
                SequentialBasedBatchAccumulator.this.notFull.signal();
                return candidate;
              } else {
                return null;
              }
            } else {
              throw new RuntimeException("Should never get to here");
            }
        }
  
      } catch (InterruptedException e) {
        LOG.error("Wait for next batch is interrupted. " + e.toString());
      } finally {
        lock.unlock();
      }
  
      return null;
    }
docs: If accumulator has been closed,  below actions are performed:
  1) remove and return the first batch if available.
  2) return null if queue is empty.
  If accumulator has not been closed, below actions are performed:
  1) if queue.size == 0, block current thread until more batches are available or accumulator is closed.
  2) if queue size == 1, remove and return the first batch if TTL has expired, else return null.
  3) if queue size > 1, remove and return the first batch element.