public static ByteBuffAllocator create(Configuration conf, boolean reservoirEnabled) {
    int poolBufSize = conf.getInt(BUFFER_SIZE_KEY, DEFAULT_BUFFER_SIZE);
    if (reservoirEnabled) {
      // The max number of buffers to be pooled in the ByteBufferPool. The default value been
      // selected based on the #handlers configured. When it is read request, 2 MB is the max size
      // at which we will send back one RPC request. Means max we need 2 MB for creating the
      // response cell block. (Well it might be much lesser than this because in 2 MB size calc, we
      // include the heap size overhead of each cells also.) Considering 2 MB, we will need
      // (2 * 1024 * 1024) / poolBufSize buffers to make the response cell block. Pool buffer size
      // is by default 64 KB.
      // In case of read request, at the end of the handler process, we will make the response
      // cellblock and add the Call to connection's response Q and a single Responder thread takes
      // connections and responses from that one by one and do the socket write. So there is chances
      // that by the time a handler originated response is actually done writing to socket and so
      // released the BBs it used, the handler might have processed one more read req. On an avg 2x
      // we consider and consider that also for the max buffers to pool
      if (poolBufSize <= 0) {
        throw new IllegalArgumentException(BUFFER_SIZE_KEY + " must be positive. Please disable "
          + "the reservoir rather than setting the size of the buffer to zero or negative.");
      }
      int bufsForTwoMB = (2 * 1024 * 1024) / poolBufSize;
      int maxBuffCount =
        conf.getInt(MAX_BUFFER_COUNT_KEY, conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
          HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT) * bufsForTwoMB * 2);
      int minSizeForReservoirUse = conf.getInt(MIN_ALLOCATE_SIZE_KEY, poolBufSize / 6);
      if (minSizeForReservoirUse <= 0) {
        LOG.warn("The minimal size for reservoir use is less or equal to zero, all allocations "
          + "will be from the pool. Set a higher " + MIN_ALLOCATE_SIZE_KEY + " to avoid this.");
      }
      Class<?> clazz = conf.getClass(BYTEBUFF_ALLOCATOR_CLASS, ByteBuffAllocator.class);
      return (ByteBuffAllocator) ReflectionUtils.newInstance(clazz, true, maxBuffCount, poolBufSize,
        minSizeForReservoirUse);
    } else {
      return HEAP;
    }
  }
Initialize an {@link ByteBuffAllocator} which will try to allocate ByteBuffers from off-heap if
reservoir is enabled and the reservoir has enough buffers, otherwise the allocator will just
allocate the insufficient buffers from on-heap to meet the requirement.
@param conf             which get the arguments to initialize the allocator.
@param reservoirEnabled indicate whether the reservoir is enabled or disabled. NOTICE: if
                        reservoir is enabled, then we will use the pool allocator to allocate
                        off-heap ByteBuffers and use the HEAP allocator to allocate heap
                        ByteBuffers. Otherwise if reservoir is disabled then all allocations
                        will happen in HEAP instance.
@return ByteBuffAllocator to manage the byte buffers.
hbase-common/src/main/java/org/apache/hadoop/hbase/io/ByteBuffAllocator.java