void doDrain(final List<RAMQueueEntry> entries, ByteBuffer metaBuff) throws InterruptedException {
    if (entries.isEmpty()) {
      return;
    }
    // This method is a little hard to follow. We run through the passed in entries and for each
    // successful add, we add a non-null BucketEntry to the below bucketEntries. Later we must
    // do cleanup making sure we've cleared ramCache of all entries regardless of whether we
    // successfully added the item to the bucketcache; if we don't do the cleanup, we'll OOME by
    // filling ramCache. We do the clean up by again running through the passed in entries
    // doing extra work when we find a non-null bucketEntries corresponding entry.
    final int size = entries.size();
    BucketEntry[] bucketEntries = new BucketEntry[size];
    // Index updated inside loop if success or if we can't succeed. We retry if cache is full
    // when we go to add an entry by going around the loop again without upping the index.
    int index = 0;
    while (isCacheEnabled() && index < size) {
      RAMQueueEntry re = null;
      try {
        re = entries.get(index);
        if (re == null) {
          LOG.warn("Couldn't get entry or changed on us; who else is messing with it?");
          index++;
          continue;
        }
        // Reset the position for reuse.
        // It should be guaranteed that the data in the metaBuff has been transferred to the
        // ioEngine safely. Otherwise, this reuse is problematic. Fortunately, the data is already
        // transferred with our current IOEngines. Should take care, when we have new kinds of
        // IOEngine in the future.
        metaBuff.clear();
        BucketEntry bucketEntry = re.writeToCache(ioEngine, bucketAllocator, realCacheSize,
          this::createRecycler, metaBuff, acceptableSize());
        // Successfully added. Up index and add bucketEntry. Clear io exceptions.
        bucketEntries[index] = bucketEntry;
        if (ioErrorStartTime > 0) {
          ioErrorStartTime = -1;
        }
        index++;
      } catch (BucketAllocatorException fle) {
        long currTs = EnvironmentEdgeManager.currentTime();
        cacheStats.allocationFailed(); // Record the warning.
        if (
          allocFailLogPrevTs == 0 || (currTs - allocFailLogPrevTs) > ALLOCATION_FAIL_LOG_TIME_PERIOD
        ) {
          LOG.warn(getAllocationFailWarningMessage(fle, re));
          allocFailLogPrevTs = currTs;
        }
        // Presume can't add. Too big? Move index on. Entry will be cleared from ramCache below.
        bucketEntries[index] = null;
        index++;
      } catch (CacheFullException cfe) {
        // Cache full when we tried to add. Try freeing space and then retrying (don't up index)
        if (!freeInProgress && !re.isPrefetch()) {
          freeSpace("Full!");
        } else if (re.isPrefetch()) {
          bucketEntries[index] = null;
          index++;
        } else {
          Thread.sleep(50);
        }
      } catch (IOException ioex) {
        // Hopefully transient. Retry. checkIOErrorIsTolerated disables cache if problem.
        LOG.error("Failed writing to bucket cache", ioex);
        checkIOErrorIsTolerated();
      }
    }

    // Make sure data pages are written on media before we update maps.
    try {
      ioEngine.sync();
    } catch (IOException ioex) {
      LOG.error("Failed syncing IO engine", ioex);
      checkIOErrorIsTolerated();
      // Since we failed sync, free the blocks in bucket allocator
      for (int i = 0; i < entries.size(); ++i) {
        BucketEntry bucketEntry = bucketEntries[i];
        if (bucketEntry != null) {
          bucketAllocator.freeBlock(bucketEntry.offset(), bucketEntry.getLength());
          bucketEntries[i] = null;
        }
      }
    }

    // Now add to backingMap if successfully added to bucket cache. Remove from ramCache if
    // success or error.
    for (int i = 0; i < size; ++i) {
      BlockCacheKey key = entries.get(i).getKey();
      // Only add if non-null entry.
      if (bucketEntries[i] != null) {
        putIntoBackingMap(key, bucketEntries[i]);
        if (ioEngine.isPersistent()) {
          setCacheInconsistent(true);
        }
      }
      // Always remove from ramCache even if we failed adding it to the block cache above.
      boolean existed = ramCache.remove(key, re -> {
        if (re != null) {
          heapSize.add(-1 * re.getData().heapSize());
        }
      });
      if (!existed && bucketEntries[i] != null) {
        // Block should have already been evicted. Remove it and free space.
        final BucketEntry bucketEntry = bucketEntries[i];
        bucketEntry.withWriteLock(offsetLock, () -> {
          if (backingMap.remove(key, bucketEntry)) {
            blockEvicted(key, bucketEntry, false, false);
          }
          return null;
        });
      }
      long used = bucketAllocator.getUsedSize();
      if (!entries.get(i).isPrefetch() && used > acceptableSize()) {
        LOG.debug("Calling freeSpace for block: {}", entries.get(i).getKey());
        freeSpace("Used=" + used + " > acceptable=" + acceptableSize());
      }
    }

  }
Flush the entries in ramCache to IOEngine and add bucket entry to backingMap. Process all that
are passed in even if failure being sure to remove from ramCache else we'll never undo the
references and we'll OOME.
@param entries Presumes list passed in here will be processed by this invocation only. No
               interference expected.
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java