HBASE-27365 Minimise block addition failures due to no space in bucke…

hbase-external-blockcache/src/main/java/org/apache/hadoop/hbase/io/hfile/MemcachedBlockCache.java

@@ -111,6 +111,11 @@ public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory)
     cacheBlock(cacheKey, buf);
   }
 
+  @Override public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
+    boolean waitWhenCache) {
+    cacheBlock(cacheKey, buf, inMemory);
+  }
+
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
   public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockCache.java

@@ -34,6 +34,16 @@ public interface BlockCache extends Iterable<CachedBlock> {
    */
   void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory);
 
+  /**
+   * Add block to cache.
+   * @param cacheKey The block's cache key.
+   * @param buf      The block contents wrapped in a ByteBuffer.
+   * @param inMemory Whether block should be treated as in-memory
+   * @param waitWhenCache Whether to wait for the cache to be flushed mainly when BucketCache
+   *                     is configured.
+   */
+  void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory, boolean waitWhenCache);
+
   /**
    * Add block to cache (defaults to not in-memory).
    * @param cacheKey The block's cache key.

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CombinedBlockCache.java

@@ -53,11 +53,16 @@ public long heapSize() {
 
   @Override
   public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory) {
+    cacheBlock(cacheKey, buf, inMemory, false);
+  }
+
+  @Override public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
+    boolean waitWhenCache) {
     boolean metaBlock = isMetaBlock(buf.getBlockType());
     if (metaBlock) {
-      l1Cache.cacheBlock(cacheKey, buf, inMemory);
+      l1Cache.cacheBlock(cacheKey, buf, inMemory, waitWhenCache);
     } else {
-      l2Cache.cacheBlock(cacheKey, buf, inMemory);
+      l2Cache.cacheBlock(cacheKey, buf, inMemory, waitWhenCache);
     }
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java

@@ -1341,8 +1341,10 @@ public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize, final bo
         // Cache the block if necessary
         cacheConf.getBlockCache().ifPresent(cache -> {
           if (cacheBlock && cacheConf.shouldCacheBlockOnRead(category)) {
+            // Using the wait on cache during compaction and prefetching.
             cache.cacheBlock(cacheKey, cacheCompressed ? hfileBlock : unpacked,
-              cacheConf.isInMemory());
+              cacheConf.isInMemory(), isCompaction ||
+                (pread  && expectedBlockType == null && expectedDataBlockEncoding == null));
           }
         });
         if (unpacked != hfileBlock) {

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java

@@ -550,7 +550,7 @@ private void doCacheOnWrite(long offset) {
       HFileBlock cacheFormatBlock = blockWriter.getBlockForCaching(cacheConf);
       try {
         cache.cacheBlock(new BlockCacheKey(name, offset, true, cacheFormatBlock.getBlockType()),
-          cacheFormatBlock);
+          cacheFormatBlock, cacheConf.isInMemory(), true);
       } finally {
         // refCnt will auto increase when block add to Cache, see RAMCache#putIfAbsent
         cacheFormatBlock.release();

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/LruAdaptiveBlockCache.java

@@ -552,6 +552,11 @@ public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory)
     }
   }
 
+  @Override public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
+    boolean waitWhenCache) {
+    cacheBlock(cacheKey, buf, inMemory);
+  }
+
   /**
    * Sanity-checking for parity between actual block cache content and metrics. Intended only for
    * use with TRACE level logging and -ea JVM.

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/LruBlockCache.java

@@ -477,6 +477,11 @@ public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
     cacheBlock(cacheKey, buf, false);
   }
 
+  @Override public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
+    boolean waitWhenCache) {
+    cacheBlock(cacheKey, buf, inMemory);
+  }
+
   /**
    * Helper function that updates the local size counter and also updates any per-cf or
    * per-blocktype metrics it can discern from given {@link LruCachedBlock}

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/TinyLfuBlockCache.java

@@ -182,6 +182,11 @@ public void cacheBlock(BlockCacheKey cacheKey, Cacheable value, boolean inMemory
     cacheBlock(cacheKey, value);
   }
 
+  @Override public void cacheBlock(BlockCacheKey cacheKey, Cacheable value, boolean inMemory,
+    boolean waitWhenCache) {
+    cacheBlock(cacheKey, value);
+  }
+
   @Override
   public void cacheBlock(BlockCacheKey key, Cacheable value) {
     if (value.heapSize() > maxBlockSize) {

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java

@@ -248,6 +248,10 @@ public class BucketCache implements BlockCache, HeapSize {
     "hbase.bucketcache.persistent.file.integrity.check.algorithm";
   private static final String DEFAULT_FILE_VERIFY_ALGORITHM = "MD5";
 
+  private static final String QUEUE_ADDITION_WAIT_TIME =
+    "hbase.bucketcache.queue.addition.waittime";
+  private static final long DEFAULT_QUEUE_ADDITION_WAIT_TIME = 0;
+  private long queueAdditionWaitTime;
   /**
    * Use {@link java.security.MessageDigest} class's encryption algorithms to check persistent file
    * integrity, default algorithm is MD5
@@ -288,6 +292,8 @@ public BucketCache(String ioEngineName, long capacity, int blockSize, int[] buck
     this.singleFactor = conf.getFloat(SINGLE_FACTOR_CONFIG_NAME, DEFAULT_SINGLE_FACTOR);
     this.multiFactor = conf.getFloat(MULTI_FACTOR_CONFIG_NAME, DEFAULT_MULTI_FACTOR);
     this.memoryFactor = conf.getFloat(MEMORY_FACTOR_CONFIG_NAME, DEFAULT_MEMORY_FACTOR);
+    this.queueAdditionWaitTime = conf.getLong(QUEUE_ADDITION_WAIT_TIME,
+      DEFAULT_QUEUE_ADDITION_WAIT_TIME);
     this.prefetchedFileListPath = conf.get(PREFETCH_PERSISTENCE_PATH_KEY);
 
     sanityCheckConfigs();
@@ -435,6 +441,18 @@ public void cacheBlock(BlockCacheKey cacheKey, Cacheable cachedItem, boolean inM
     cacheBlockWithWait(cacheKey, cachedItem, inMemory, wait_when_cache);
   }
 
+  /**
+   * Cache the block with the specified name and buffer.
+   * @param cacheKey   block's cache key
+   * @param cachedItem block buffer
+   * @param inMemory   if block is in-memory
+   */
+  @Override
+  public void cacheBlock(BlockCacheKey cacheKey, Cacheable cachedItem, boolean inMemory,
+    boolean waitWhenCache) {
+    cacheBlockWithWait(cacheKey, cachedItem, inMemory, waitWhenCache && queueAdditionWaitTime > 0);
+  }
+
   /**
    * Cache the block to ramCache
    * @param cacheKey   block's cache key
@@ -491,7 +509,7 @@ protected void cacheBlockWithWaitInternal(BlockCacheKey cacheKey, Cacheable cach
     boolean successfulAddition = false;
     if (wait) {
       try {
-        successfulAddition = bq.offer(re, DEFAULT_CACHE_WAIT_TIME, TimeUnit.MILLISECONDS);
+        successfulAddition = bq.offer(re,  queueAdditionWaitTime, TimeUnit.MILLISECONDS);
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
       }

hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java

@@ -58,6 +58,7 @@
 import org.apache.hadoop.hbase.testclassification.IOTests;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.hbase.util.Threads;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -811,4 +812,56 @@ public void testFreeBucketEntryRestoredFromFile() throws Exception {
     }
   }
 
+  /**
+   * This test is for HBASE-26295, {@link BucketEntry} which is restored from a persistence file
+   * could not be freed even if corresponding {@link HFileBlock} is evicted from
+   * {@link BucketCache}.
+   */
+  @Test
+  public void testBlockAdditionWaitWhenCache() throws Exception {
+    try {
+      final Path dataTestDir = createAndGetTestDir();
+
+      String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache";
+      String persistencePath = dataTestDir + "/bucketNoRecycler.persistence";
+
+      BucketCache bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
+        constructedBlockSizes, 1, 1, persistencePath);
+      long usedByteSize = bucketCache.getAllocator().getUsedSize();
+      assertEquals(0, usedByteSize);
+
+      HFileBlockPair[] hfileBlockPairs =
+        CacheTestUtils.generateHFileBlocks(constructedBlockSize, 10);
+      // Add blocks
+      for (HFileBlockPair hfileBlockPair : hfileBlockPairs) {
+        bucketCache.cacheBlock(hfileBlockPair.getBlockName(), hfileBlockPair.getBlock(),
+          false, true);
+      }
+
+      // Wait for blocks size to match the number of blocks.
+      while(bucketCache.backingMap.size() != 10) {
+        Threads.sleep(100);
+      }
+      usedByteSize = bucketCache.getAllocator().getUsedSize();
+      assertNotEquals(0, usedByteSize);
+      // persist cache to file
+      bucketCache.shutdown();
+      assertTrue(new File(persistencePath).exists());
+
+      // restore cache from file
+      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
+        constructedBlockSizes, writeThreads, writerQLen, persistencePath);
+      assertFalse(new File(persistencePath).exists());
+      assertEquals(usedByteSize, bucketCache.getAllocator().getUsedSize());
+
+      for (HFileBlockPair hfileBlockPair : hfileBlockPairs) {
+        BlockCacheKey blockCacheKey = hfileBlockPair.getBlockName();
+        bucketCache.evictBlock(blockCacheKey);
+      }
+      assertEquals(0, bucketCache.getAllocator().getUsedSize());
+      assertEquals(0, bucketCache.backingMap.size());
+    } finally {
+      HBASE_TESTING_UTILITY.cleanupTestDir();
+    }
+  }
 }

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHeapMemoryManager.java

@@ -651,6 +651,11 @@ public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
 
     }
 
+    @Override public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
+      boolean waitWhenCache) {
+
+    }
+
     @Override
     public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching, boolean repeat,
       boolean updateCacheMetrics) {

hbase-server/src/test/resources/hbase-site.xml

@@ -277,4 +277,9 @@
     <value>3</value>
     <description>Default is unbounded</description>
   </property>
+  <property>
+    <name>hbase.bucketcache.queue.addition.waittime</name>
+    <value>100</value>
+    <description>Default is 0</description>
+  </property>
 </configuration>