@Override
  public void notifyFileCachingCompleted(Path fileName, int totalBlockCount, int dataBlockCount,
    long size) {
    // block eviction may be happening in the background as prefetch runs,
    // so we need to count all blocks for this file in the backing map under
    // a read lock for the block offset
    final List<ReentrantReadWriteLock> locks = new ArrayList<>();
    LOG.debug("Notifying caching completed for file {}, with total blocks {}, and data blocks {}",
      fileName, totalBlockCount, dataBlockCount);
    try {
      final MutableInt count = new MutableInt();
      LOG.debug("iterating over {} entries in the backing map", backingMap.size());
      Set<BlockCacheKey> result = getAllCacheKeysForFile(fileName.getName(), 0, Long.MAX_VALUE);
      if (result.isEmpty() && StoreFileInfo.isReference(fileName)) {
        result = getAllCacheKeysForFile(
          StoreFileInfo.getReferredToRegionAndFile(fileName.getName()).getSecond(), 0,
          Long.MAX_VALUE);
      }
      result.stream().forEach(entry -> {
        LOG.debug("found block for file {} in the backing map. Acquiring read lock for offset {}",
          fileName.getName(), entry.getOffset());
        ReentrantReadWriteLock lock = offsetLock.getLock(entry.getOffset());
        lock.readLock().lock();
        locks.add(lock);
        if (backingMap.containsKey(entry) && entry.getBlockType().isData()) {
          count.increment();
        }
      });
      // BucketCache would only have data blocks
      if (dataBlockCount == count.getValue()) {
        LOG.debug("File {} has now been fully cached.", fileName);
        fileCacheCompleted(fileName, size);
      } else {
        LOG.debug(
          "Prefetch executor completed for {}, but only {} data blocks were cached. "
            + "Total data blocks for file: {}. "
            + "Checking for blocks pending cache in cache writer queue.",
          fileName, count.getValue(), dataBlockCount);
        if (ramCache.hasBlocksForFile(fileName.getName())) {
          for (ReentrantReadWriteLock lock : locks) {
            lock.readLock().unlock();
          }
          locks.clear();
          LOG.debug("There are still blocks pending caching for file {}. Will sleep 100ms "
            + "and try the verification again.", fileName);
          Thread.sleep(100);
          notifyFileCachingCompleted(fileName, totalBlockCount, dataBlockCount, size);
        } else {
          LOG.info(
            "The total block count was {}. We found only {} data blocks cached from "
              + "a total of {} data blocks for file {}, "
              + "but no blocks pending caching. Maybe cache is full or evictions "
              + "happened concurrently to cache prefetch.",
            totalBlockCount, count, dataBlockCount, fileName);
        }
      }
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    } finally {
      for (ReentrantReadWriteLock lock : locks) {
        lock.readLock().unlock();
      }
    }
  }
Defined an {@link Consumer} here, because once the removed entry release its reference count,
then it's ByteBuffers may be recycled and accessing it outside this method will be thrown an
exception. the consumer will access entry to remove before release its reference count.
Notice, don't change its reference count in the {@link Consumer}
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java