class BatchingReadPipe implements ReadPipe<List<T>> {

  @Override
  public List<T> poll() {
    final List<T> batch = new ArrayList<>();
    input.drainTo(batch, maximumBatchSize);
    if (batch.isEmpty()) {
      // Poll has to return null if the pipe is empty
      return null;
    }
    batchCounter.inc();
    return batch;
  }

  public  BatchingReadPipe(
      final ReadPipe<T> input, final int maximumBatchSize, final Counter batchCounter);

}