diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java index 19cf2af56ee2..6cd632a33bfb 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java @@ -92,6 +92,7 @@ import org.apache.hadoop.hbase.util.IdReadWriteLockWithObjectPool; import org.apache.hadoop.hbase.util.IdReadWriteLockWithObjectPool.ReferenceType; import org.apache.hadoop.hbase.util.Pair; +import org.apache.hadoop.hbase.util.Threads; import org.apache.hadoop.util.StringUtils; import org.apache.yetus.audience.InterfaceAudience; import org.slf4j.Logger; @@ -206,6 +207,9 @@ protected enum CacheState { */ private volatile CacheState cacheState; + /** The single cleanup thread shared by disable and explicit shutdown calls. */ + private volatile Thread cacheCleanupThread; + /** * A list of writer queues. We have a queue per {@link WriterThread} we have running. In other * words, the work adding blocks to the BucketCache is divided up amongst the running @@ -690,6 +694,7 @@ public Cacheable getBlock(BlockCacheKey key, boolean caching, boolean repeat, if (bucketEntry != null) { long start = System.nanoTime(); ReentrantReadWriteLock lock = offsetLock.getLock(bucketEntry.offset()); + boolean failedBucketEntryRead = false; try { lock.readLock().lock(); // We can not read here even if backingMap does contain the given key because its offset @@ -722,8 +727,7 @@ public Cacheable getBlock(BlockCacheKey key, boolean caching, boolean repeat, // When using file io engine persistent cache, // the cache map state might differ from the actual cache. If we reach this block, // we should remove the cache key entry from the backing map - backingMap.remove(key); - fileNotFullyCached(key, bucketEntry); + failedBucketEntryRead = true; LOG.debug("Failed to fetch block for cache key: {}.", key, hioex); } catch (IOException ioex) { LOG.error("Failed reading block " + key + " from bucket cache", ioex); @@ -731,6 +735,9 @@ public Cacheable getBlock(BlockCacheKey key, boolean caching, boolean repeat, } finally { lock.readLock().unlock(); } + if (failedBucketEntryRead) { + removeFailedBucketEntry(bucketEntry); + } } if (!repeat && updateCacheMetrics) { cacheStats.miss(caching, key.isPrimary(), key.getBlockType()); @@ -738,6 +745,30 @@ public Cacheable getBlock(BlockCacheKey key, boolean caching, boolean repeat, return null; } + private void removeFailedBucketEntry(BucketEntry bucketEntry) { + BlockCacheKey cacheKey = findBackingMapKey(bucketEntry); + if (cacheKey == null) { + return; + } + bucketEntry.withWriteLock(offsetLock, () -> { + if (backingMap.remove(cacheKey, bucketEntry)) { + blockEvicted(cacheKey, bucketEntry, true, false); + } + return null; + }); + } + + private BlockCacheKey findBackingMapKey(BucketEntry bucketEntry) { + // Reference file lookups use a different key from the one stored in backingMap. This only runs + // after a failed read, so find the stored key to preserve its region metadata during eviction. + for (Map.Entry entry : backingMap.entrySet()) { + if (entry.getValue() == bucketEntry) { + return entry.getKey(); + } + } + return null; + } + /** * This method is invoked after the bucketEntry is removed from {@link BucketCache#backingMap} */ @@ -1798,7 +1829,7 @@ private void checkIOErrorIsTolerated() { if (isCacheEnabled() && (now - ioErrorStartTimeTmp) > this.ioErrorsTolerationDuration) { LOG.error("IO errors duration time has exceeded " + ioErrorsTolerationDuration + "ms, disabling cache, please check your IOEngine"); - disableCache(); + disableCache(false); } } else { this.ioErrorStartTime = now; @@ -1806,9 +1837,11 @@ private void checkIOErrorIsTolerated() { } /** - * Used to shut down the cache -or- turn it off in the case of something broken. + * Used to shut down the cache -or- turn it off in the case of something broken. Cleanup runs + * separately because IO errors can invoke this method from a writer thread or while holding an + * offset read lock. */ - private void disableCache() { + private synchronized void disableCache(boolean persistOnCleanup) { if (!isCacheEnabled()) { return; } @@ -1816,52 +1849,90 @@ private void disableCache() { cacheState = CacheState.DISABLED; ioEngine.shutdown(); this.scheduleThreadPool.shutdown(); - for (int i = 0; i < writerThreads.length; ++i) - writerThreads[i].interrupt(); - this.ramCache.clear(); - if (!ioEngine.isPersistent() || persistencePath == null) { - // If persistent ioengine and a path, we will serialize out the backingMap. - this.backingMap.clear(); - this.blocksByHFile.clear(); - this.fullyCachedFiles.clear(); - this.regionCachedSize.clear(); + for (WriterThread writerThread : writerThreads) { + writerThread.interrupt(); } + ramCache.clear(); if (cacheStats.getMetricsRollerScheduler() != null) { cacheStats.getMetricsRollerScheduler().shutdownNow(); } + cacheCleanupThread = + Threads.setDaemonThreadRunning(new Thread(() -> cleanupCache(persistOnCleanup)), + "BucketCacheCleanup-" + System.identityHashCode(this), Threads.LOGGING_EXCEPTION_HANDLER); } - private void join() throws InterruptedException { - for (int i = 0; i < writerThreads.length; ++i) - writerThreads[i].join(); - } - - @Override - public void shutdown() { - if (isCacheEnabled()) { - disableCache(); - LOG.info("Shutdown bucket cache: IO persistent=" + ioEngine.isPersistent() - + "; path to write=" + persistencePath); - if (ioEngine.isPersistent() && persistencePath != null) { + private void cleanupCache(boolean persistOnCleanup) { + try { + joinWriterThreads(); + stopCachePersister(); + if (persistOnCleanup && isCachePersistent()) { try { - join(); - if (cachePersister != null) { - LOG.info("Shutting down cache persister thread."); - cachePersister.shutdown(); - while (cachePersister.isAlive()) { - Thread.sleep(10); - } - } persistToFile(); } catch (IOException ex) { LOG.error("Unable to persist data on exit: " + ex.toString(), ex); - } catch (InterruptedException e) { - LOG.warn("Failed to persist data on exit", e); } } + } finally { + releaseBackingMapReferences(); + } + } + + private void joinWriterThreads() { + for (WriterThread writerThread : writerThreads) { + Threads.shutdown(writerThread); + } + } + + private void stopCachePersister() { + if (cachePersister != null) { + LOG.info("Shutting down cache persister thread."); + cachePersister.shutdown(); + Threads.shutdown(cachePersister); + } + } + + private void releaseBackingMapReferences() { + for (Map.Entry entry : backingMap.entrySet()) { + BlockCacheKey cacheKey = entry.getKey(); + BucketEntry bucketEntry = entry.getValue(); + bucketEntry.withWriteLock(offsetLock, () -> { + if (backingMap.remove(cacheKey, bucketEntry)) { + bucketEntry.markAsEvicted(); + } + return null; + }); + } + blocksByHFile.clear(); + fullyCachedFiles.clear(); + regionCachedSize.clear(); + } + + private void waitForCacheCleanup() { + Thread cleanupThread = cacheCleanupThread; + if (cleanupThread == null || cleanupThread == Thread.currentThread()) { + return; + } + boolean interrupted = false; + while (cleanupThread.isAlive()) { + try { + cleanupThread.join(); + } catch (InterruptedException e) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); } } + @Override + public void shutdown() { + disableCache(true); + waitForCacheCleanup(); + LOG.info("Shutdown bucket cache: IO persistent=" + ioEngine.isPersistent() + "; path to write=" + + persistencePath); + } + /** * Needed mostly for UTs that might run in the same VM and create different BucketCache instances * on different UT methods. diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockEvictionOnRegionMovement.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockEvictionOnRegionMovement.java index cb3f7e5772be..b70f65a59514 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockEvictionOnRegionMovement.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockEvictionOnRegionMovement.java @@ -27,6 +27,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseTestingUtil; +import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.SingleProcessHBaseCluster; import org.apache.hadoop.hbase.StartTestingClusterOption; import org.apache.hadoop.hbase.TableName; @@ -126,20 +127,26 @@ public void testBlockEvictionOnGracefulStop() throws Exception { : cluster.getRegionServer(0); assertTrue(regionServingRS.getBlockCache().isPresent()); - long oldUsedCacheSize = - regionServingRS.getBlockCache().get().getBlockCaches()[1].getCurrentSize(); - assertNotEquals(0, regionServingRS.getBlockCache().get().getBlockCaches()[1].getBlockCount()); - - cluster.stopRegionServer(regionServingRS.getServerName()); - Thread.sleep(500); - cluster.startRegionServer(); - Thread.sleep(500); - - regionServingRS.getBlockCache().get().waitForCacheInitialization(10000); - long newUsedCacheSize = - regionServingRS.getBlockCache().get().getBlockCaches()[1].getCurrentSize(); - assertEquals(oldUsedCacheSize, newUsedCacheSize); - assertNotEquals(0, regionServingRS.getBlockCache().get().getBlockCaches()[1].getBlockCount()); + BlockCache oldBucketCache = regionServingRS.getBlockCache().get().getBlockCaches()[1]; + long oldUsedCacheSize = oldBucketCache.getCurrentSize(); + assertNotEquals(0, oldUsedCacheSize); + assertNotEquals(0, oldBucketCache.getBlockCount()); + + ServerName serverName = regionServingRS.getServerName(); + cluster.stopRegionServer(serverName); + cluster.waitForRegionServerToStop(serverName, 10000); + + // Persistent state is restored into a new cache; the stopped cache must release its + // references. + assertEquals(0, oldBucketCache.getCurrentSize()); + + HRegionServer restartedRegionServer = cluster.startRegionServer().getRegionServer(); + assertTrue(restartedRegionServer.getBlockCache().isPresent()); + BlockCache restoredBucketCache = + restartedRegionServer.getBlockCache().get().getBlockCaches()[1]; + assertTrue(restoredBucketCache.waitForCacheInitialization(10000)); + assertEquals(oldUsedCacheSize, restoredBucketCache.getCurrentSize()); + assertNotEquals(0, restoredBucketCache.getBlockCount()); } public TableName writeDataToTable(String testName) throws IOException, InterruptedException { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java index 6a5eac5fe3d7..4d9bd2dfcd91 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java @@ -841,10 +841,15 @@ public void testFreeBucketEntryRestoredFromFile() throws Exception { cacheAndWaitUntilFlushedToBucket(bucketCache, hfileBlockPair.getBlockName(), hfileBlockPair.getBlock(), false); } + BucketEntry bucketEntryBeforeShutdown = + bucketCache.backingMap.get(hfileBlockPairs[0].getBlockName()); + assertNotNull(bucketEntryBeforeShutdown); + assertEquals(1, bucketEntryBeforeShutdown.refCnt()); usedByteSize = bucketCache.getAllocator().getUsedSize(); assertNotEquals(0, usedByteSize); // persist cache to file bucketCache.shutdown(); + assertEquals(0, bucketEntryBeforeShutdown.refCnt()); assertTrue(new File(persistencePath).exists()); // restore cache from file diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCacheRefCnt.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCacheRefCnt.java index b466719136c0..19fc42ebf92c 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCacheRefCnt.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCacheRefCnt.java @@ -23,16 +23,26 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.atomic.LongAdder; +import java.util.function.Function; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.Waiter; import org.apache.hadoop.hbase.io.ByteBuffAllocator; +import org.apache.hadoop.hbase.io.ByteBuffAllocator.Recycler; import org.apache.hadoop.hbase.io.hfile.BlockCacheKey; import org.apache.hadoop.hbase.io.hfile.BlockCacheUtil; import org.apache.hadoop.hbase.io.hfile.BlockType; @@ -40,6 +50,7 @@ import org.apache.hadoop.hbase.io.hfile.HFileBlock; import org.apache.hadoop.hbase.io.hfile.HFileContext; import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder; +import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.RAMQueueEntry; import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.WriterThread; import org.apache.hadoop.hbase.nio.ByteBuff; import org.apache.hadoop.hbase.nio.RefCnt; @@ -48,6 +59,9 @@ import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.apache.hbase.thirdparty.com.google.common.util.concurrent.Uninterruptibles; @Tag(IOTests.TAG) @Tag(SmallTests.TAG) @@ -306,6 +320,261 @@ public void testMarkStaleAsEvicted() throws Exception { } } + @Test + public void testShutdownReleasesBackingMapReferenceWhileCallerRetainsBlock() throws Exception { + ByteBuffAllocator alloc = ByteBuffAllocator.create(HBaseConfiguration.create(), true); + HFileBlock blockToCache = createBlock(200, 1020, alloc); + HFileBlock blockFromCache = null; + try { + cache = create(1, 1000); + BlockCacheKey key = + createKey("testShutdownReleasesBackingMapReferenceWhileCallerRetainsBlock", 200); + cache.cacheBlock(key, blockToCache); + waitUntilFlushedToCache(cache, key); + + blockFromCache = (HFileBlock) cache.getBlock(key, false, false, false); + assertNotNull(blockFromCache); + assertEquals(2, blockFromCache.refCnt()); + + cache.shutdown(); + cache = null; + + assertEquals(1, blockFromCache.refCnt(), + "shutdown must release only the reference owned by backingMap"); + assertTrue(blockFromCache.release()); + assertEquals(0, blockFromCache.refCnt()); + } finally { + if (cache != null) { + cache.shutdown(); + cache = null; + } + if (blockFromCache != null) { + while (blockFromCache.refCnt() > 0) { + blockFromCache.release(); + } + } + while (blockToCache.refCnt() > 0) { + blockToCache.release(); + } + alloc.clean(); + } + } + + @Test + public void testHBaseIOExceptionReleasesBackingMapReference(@TempDir File testDir) + throws Exception { + ByteBuffAllocator alloc = ByteBuffAllocator.create(HBaseConfiguration.create(), true); + HFileBlock blockToCache = createBlock(200, 1020, alloc); + BucketEntry bucketEntry = null; + String cachePath = new File(testDir, "bucket.cache").getAbsolutePath(); + String persistencePath = new File(testDir, "bucket.persistence").getAbsolutePath(); + BucketCache bucketCache = new BucketCache("file:" + cachePath, CAPACITY_SIZE, BLOCK_SIZE, + BLOCK_SIZE_ARRAY, 1, 1000, persistencePath); + try { + assertTrue(bucketCache.waitForCacheInitialization(10000)); + BlockCacheKey key = createKey("testHBaseIOExceptionReleasesBackingMapReference", 200); + bucketCache.cacheBlock(key, blockToCache); + waitUntilFlushedToCache(bucketCache, key); + + bucketEntry = bucketCache.backingMap.get(key); + assertNotNull(bucketEntry); + assertEquals(1, bucketEntry.refCnt()); + + ByteBuffer invalidCachedTime = ByteBuffer.allocate(Long.BYTES); + invalidCachedTime.putLong(bucketEntry.getCachedTime() + 1).flip(); + bucketCache.ioEngine.write(invalidCachedTime, bucketEntry.offset()); + bucketCache.ioEngine.sync(); + + assertNull(bucketCache.getBlock(key, false, false, false)); + assertFalse(bucketCache.backingMap.containsKey(key)); + assertEquals(0, bucketEntry.refCnt(), + "removing a corrupt entry must release the reference owned by backingMap"); + assertEquals(0, bucketCache.getAllocator().getUsedSize()); + } finally { + bucketCache.shutdown(); + if (bucketEntry != null && bucketEntry.refCnt() > 0) { + bucketEntry.markAsEvicted(); + } + while (blockToCache.refCnt() > 0) { + blockToCache.release(); + } + alloc.clean(); + } + } + + @Test + public void testHBaseIOExceptionThroughReferenceReleasesBackingMapReference(@TempDir File testDir) + throws Exception { + ByteBuffAllocator alloc = ByteBuffAllocator.create(HBaseConfiguration.create(), true); + HFileBlock blockToCache = createBlock(200, 1020, alloc); + BucketEntry bucketEntry = null; + String cachePath = new File(testDir, "bucket.cache").getAbsolutePath(); + String persistencePath = new File(testDir, "bucket.persistence").getAbsolutePath(); + BucketCache bucketCache = new BucketCache("file:" + cachePath, CAPACITY_SIZE, BLOCK_SIZE, + BLOCK_SIZE_ARRAY, 1, 1000, persistencePath); + try { + assertTrue(bucketCache.waitForCacheInitialization(10000)); + String hfileName = "0123456789abcdef"; + String regionName = "region"; + BlockCacheKey storedKey = + new BlockCacheKey(hfileName, "cf", regionName, 200, true, BlockType.DATA, false); + BlockCacheKey referenceKey = createKey(hfileName + ".parent", 200); + bucketCache.cacheBlock(storedKey, blockToCache); + waitUntilFlushedToCache(bucketCache, storedKey); + + bucketEntry = bucketCache.backingMap.get(storedKey); + assertNotNull(bucketEntry); + assertEquals(1, bucketEntry.refCnt()); + assertTrue(bucketCache.regionCachedSize.containsKey(regionName)); + + ByteBuffer invalidCachedTime = ByteBuffer.allocate(Long.BYTES); + invalidCachedTime.putLong(bucketEntry.getCachedTime() + 1).flip(); + bucketCache.ioEngine.write(invalidCachedTime, bucketEntry.offset()); + bucketCache.ioEngine.sync(); + + assertNull(bucketCache.getBlock(referenceKey, false, false, false)); + assertFalse(bucketCache.backingMap.containsKey(storedKey)); + assertEquals(0, bucketEntry.refCnt(), + "removing a corrupt referred entry must release the reference owned by backingMap"); + assertFalse(bucketCache.regionCachedSize.containsKey(regionName)); + assertEquals(0, bucketCache.getAllocator().getUsedSize()); + } finally { + bucketCache.shutdown(); + if (bucketEntry != null && bucketEntry.refCnt() > 0) { + bucketEntry.markAsEvicted(); + } + while (blockToCache.refCnt() > 0) { + blockToCache.release(); + } + alloc.clean(); + } + } + + @Test + public void testIoErrorDisableReleasesBackingMapReference() throws Exception { + ByteBuffAllocator alloc = ByteBuffAllocator.create(HBaseConfiguration.create(), true); + HFileBlock blockToCache = createBlock(200, 1020, alloc); + BucketCache bucketCache = new BucketCache(IO_ENGINE, CAPACITY_SIZE, BLOCK_SIZE, + BLOCK_SIZE_ARRAY, 1, 1000, PERSISTENCE_PATH, 0, HBaseConfiguration.create()); + try { + BlockCacheKey key = createKey("testIoErrorDisableReleasesBackingMapReference", 200); + bucketCache.cacheBlock(key, blockToCache); + waitUntilFlushedToCache(bucketCache, key); + + BucketEntry bucketEntry = bucketCache.backingMap.get(key); + assertNotNull(bucketEntry); + assertEquals(1, bucketEntry.refCnt()); + + BlockCacheKey failingKey = createKey("failing", 400); + RAMQueueEntry failingEntry = new FailingRAMQueueEntry(failingKey, blockToCache); + bucketCache.doDrain(Arrays.asList(failingEntry), + ByteBuffer.allocate(HFileBlock.BLOCK_METADATA_SPACE)); + assertFalse(bucketCache.isCacheEnabled()); + + bucketCache.shutdown(); + assertEquals(0, bucketEntry.refCnt()); + assertTrue(bucketCache.backingMap.isEmpty()); + } finally { + bucketCache.shutdown(); + while (blockToCache.refCnt() > 0) { + blockToCache.release(); + } + alloc.clean(); + } + } + + @Test + public void testShutdownCleansEntryAddedByInFlightWriter() throws Exception { + ByteBuffAllocator alloc = ByteBuffAllocator.create(HBaseConfiguration.create(), true); + HFileBlock blockToCache = createBlock(200, 1020, alloc); + PausingPutBucketCache bucketCache = new PausingPutBucketCache(IO_ENGINE, CAPACITY_SIZE, + BLOCK_SIZE, BLOCK_SIZE_ARRAY, 1, 1000, PERSISTENCE_PATH); + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future shutdownFuture = null; + try { + BlockCacheKey key = createKey("testShutdownCleansEntryAddedByInFlightWriter", 200); + bucketCache.cacheBlock(key, blockToCache); + assertTrue(bucketCache.entryAdded.await(10, TimeUnit.SECONDS)); + + BucketEntry bucketEntry = bucketCache.addedEntry.get(); + assertNotNull(bucketEntry); + assertEquals(1, bucketEntry.refCnt()); + + shutdownFuture = executor.submit(bucketCache::shutdown); + Waiter.waitFor(HBaseConfiguration.create(), 10000, () -> !bucketCache.isCacheEnabled()); + assertFalse(shutdownFuture.isDone(), + "shutdown must wait for a writer that can still update backingMap"); + + bucketCache.continueWriter.countDown(); + shutdownFuture.get(10, TimeUnit.SECONDS); + + assertFalse(bucketCache.writerThreads[0].isAlive()); + assertTrue(bucketCache.backingMap.isEmpty()); + assertTrue(bucketCache.ramCache.isEmpty()); + assertEquals(0, bucketEntry.refCnt()); + } finally { + bucketCache.continueWriter.countDown(); + bucketCache.shutdown(); + bucketCache.writerThreads[0].join(TimeUnit.SECONDS.toMillis(10)); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + BucketEntry addedEntry = bucketCache.addedEntry.get(); + if (addedEntry != null && addedEntry.refCnt() > 0) { + addedEntry.markAsEvicted(); + } + while (blockToCache.refCnt() > 0) { + blockToCache.release(); + } + alloc.clean(); + } + } + + @Test + public void testConcurrentAndRepeatedShutdownReleaseBackingMapReferenceOnce() throws Exception { + ByteBuffAllocator alloc = ByteBuffAllocator.create(HBaseConfiguration.create(), true); + HFileBlock blockToCache = createBlock(200, 1020, alloc); + BlockingCleanupBucketCache bucketCache = new BlockingCleanupBucketCache(IO_ENGINE, + CAPACITY_SIZE, BLOCK_SIZE, BLOCK_SIZE_ARRAY, 1, 1000, PERSISTENCE_PATH); + ExecutorService executor = Executors.newFixedThreadPool(2); + BucketEntry bucketEntry = null; + try { + BlockCacheKey key = + createKey("testConcurrentAndRepeatedShutdownReleaseBackingMapReferenceOnce", 200); + bucketCache.cacheBlock(key, blockToCache); + waitUntilFlushedToCache(bucketCache, key); + bucketEntry = bucketCache.backingMap.get(key); + assertNotNull(bucketEntry); + assertEquals(1, bucketEntry.refCnt()); + + Future firstShutdown = executor.submit(bucketCache::shutdown); + assertTrue(bucketCache.firstFree.await(10, TimeUnit.SECONDS)); + + Future secondShutdown = executor.submit(bucketCache::shutdown); + assertTrue(bucketCache.secondShutdownStarted.await(10, TimeUnit.SECONDS)); + bucketCache.continueFree.countDown(); + + firstShutdown.get(10, TimeUnit.SECONDS); + secondShutdown.get(10, TimeUnit.SECONDS); + bucketCache.shutdown(); + + assertEquals(1, bucketCache.freeBucketEntryCount.get()); + assertEquals(0, bucketEntry.refCnt()); + assertTrue(bucketCache.backingMap.isEmpty()); + } finally { + bucketCache.continueFree.countDown(); + bucketCache.shutdown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + if (bucketEntry != null && bucketEntry.refCnt() > 0) { + bucketEntry.markAsEvicted(); + } + while (blockToCache.refCnt() > 0) { + blockToCache.release(); + } + alloc.clean(); + } + } + /** * 
    * This test is for HBASE-26281,
@@ -485,6 +754,79 @@ public void testEvictingBlockCachingBlockGettingBlockConcurrently() throws Excep
 
   }
 
+  private static final class PausingPutBucketCache extends BucketCache {
+    private final CountDownLatch entryAdded = new CountDownLatch(1);
+    private final CountDownLatch continueWriter = new CountDownLatch(1);
+    private final AtomicReference addedEntry = new AtomicReference<>();
+
+    private PausingPutBucketCache(String ioEngineName, long capacity, int blockSize,
+      int[] bucketSizes, int writerThreadNum, int writerQLen, String persistencePath)
+      throws IOException {
+      super(ioEngineName, capacity, blockSize, bucketSizes, writerThreadNum, writerQLen,
+        persistencePath);
+    }
+
+    @Override
+    protected void putIntoBackingMap(BlockCacheKey key, BucketEntry bucketEntry) {
+      super.putIntoBackingMap(key, bucketEntry);
+      addedEntry.set(bucketEntry);
+      entryAdded.countDown();
+      if (!Uninterruptibles.awaitUninterruptibly(continueWriter, 10, TimeUnit.SECONDS)) {
+        throw new AssertionError("Timed out waiting to resume the bucket cache writer");
+      }
+    }
+  }
+
+  private static final class FailingRAMQueueEntry extends RAMQueueEntry {
+    private FailingRAMQueueEntry(BlockCacheKey key, Cacheable data) {
+      super(key, data, 0, false, false, false);
+    }
+
+    @Override
+    public BucketEntry writeToCache(IOEngine ioEngine, BucketAllocator alloc,
+      LongAdder realCacheSize, Function createRecycler, ByteBuffer metaBuff,
+      Long acceptableSize) throws IOException {
+      Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS);
+      throw new IOException("Mocked!");
+    }
+  }
+
+  private static final class BlockingCleanupBucketCache extends BucketCache {
+    private final CountDownLatch firstFree = new CountDownLatch(1);
+    private final CountDownLatch continueFree = new CountDownLatch(1);
+    private final CountDownLatch secondShutdownStarted = new CountDownLatch(1);
+    private final AtomicBoolean blockFirstFree = new AtomicBoolean(true);
+    private final AtomicInteger freeBucketEntryCount = new AtomicInteger();
+    private final AtomicInteger shutdownCount = new AtomicInteger();
+
+    private BlockingCleanupBucketCache(String ioEngineName, long capacity, int blockSize,
+      int[] bucketSizes, int writerThreadNum, int writerQLen, String persistencePath)
+      throws IOException {
+      super(ioEngineName, capacity, blockSize, bucketSizes, writerThreadNum, writerQLen,
+        persistencePath);
+    }
+
+    @Override
+    public void shutdown() {
+      if (shutdownCount.incrementAndGet() == 2) {
+        secondShutdownStarted.countDown();
+      }
+      super.shutdown();
+    }
+
+    @Override
+    void freeBucketEntry(BucketEntry bucketEntry) {
+      freeBucketEntryCount.incrementAndGet();
+      if (blockFirstFree.compareAndSet(true, false)) {
+        firstFree.countDown();
+        if (!Uninterruptibles.awaitUninterruptibly(continueFree, 10, TimeUnit.SECONDS)) {
+          throw new AssertionError("Timed out waiting to resume backingMap cleanup");
+        }
+      }
+      super.freeBucketEntry(bucketEntry);
+    }
+  }
+
   static class MyBucketCache extends BucketCache {
     private static final String GET_BLOCK_THREAD_NAME = "_getBlockThread";
     private static final String CACHE_BLOCK_THREAD_NAME = "_cacheBlockThread";