@Test
  public void testReplacingBlockAndGettingBlockConcurrently() throws Exception {
    ByteBuffAllocator byteBuffAllocator =
      ByteBuffAllocator.create(HBaseConfiguration.create(), true);
    final MyBucketCache myBucketCache = createMyBucketCache(1, 1000);
    try {
      HFileBlock hfileBlock = createBlock(200, 1020, byteBuffAllocator);
      final BlockCacheKey blockCacheKey = createKey("testTwoThreadConcurrent", 200);
      myBucketCache.cacheBlock(blockCacheKey, hfileBlock);
      waitUntilFlushedToCache(myBucketCache, blockCacheKey);
      assertEquals(1, hfileBlock.refCnt());

      assertTrue(!myBucketCache.ramCache.containsKey(blockCacheKey));
      final AtomicReference<Throwable> exceptionRef = new AtomicReference<Throwable>();
      Thread cacheBlockThread = new Thread(() -> {
        try {
          HFileBlock newHFileBlock = createBlock(200, 1020, byteBuffAllocator);
          myBucketCache.cacheBlock(blockCacheKey, newHFileBlock);
          waitUntilFlushedToCache(myBucketCache, blockCacheKey);

        } catch (Throwable exception) {
          exceptionRef.set(exception);
        }
      });
      cacheBlockThread.setName(MyBucketCache.CACHE_BLOCK_THREAD_NAME);
      cacheBlockThread.start();

      String oldThreadName = Thread.currentThread().getName();
      HFileBlock gotHFileBlock = null;
      try {

        Thread.currentThread().setName(MyBucketCache.GET_BLOCK_THREAD_NAME);

        gotHFileBlock = (HFileBlock) (myBucketCache.getBlock(blockCacheKey, false, false, false));
        assertTrue(gotHFileBlock.equals(hfileBlock));
        assertTrue(gotHFileBlock.getByteBuffAllocator() == byteBuffAllocator);
        assertEquals(2, gotHFileBlock.refCnt());
        /**
         * Release the second cyclicBarrier.await in
         * {@link MyBucketCache#cacheBlockWithWaitInternal}
         */
        myBucketCache.cyclicBarrier.await();

      } finally {
        Thread.currentThread().setName(oldThreadName);
      }

      cacheBlockThread.join();
      assertTrue(exceptionRef.get() == null);
      assertEquals(1, gotHFileBlock.refCnt());
      assertTrue(gotHFileBlock.equals(hfileBlock));
      assertTrue(myBucketCache.overwiteByteBuff == null);
      assertTrue(myBucketCache.freeBucketEntryCounter.get() == 0);

      gotHFileBlock.release();
      assertEquals(0, gotHFileBlock.refCnt());
      assertTrue(myBucketCache.overwiteByteBuff != null);
      assertTrue(myBucketCache.freeBucketEntryCounter.get() == 1);
      assertTrue(myBucketCache.replaceCounter.get() == 1);
      assertTrue(myBucketCache.blockEvictCounter.get() == 1);
    } finally {
      myBucketCache.shutdown();
    }

  }
<pre>
This test is for HBASE-26281,
test two threads for replacing Block and getting Block execute concurrently.
The threads sequence is:
1. Block1 was cached successfully,the {@link RefCnt} of Block1 is 1.
2. Thread1 caching the same {@link BlockCacheKey} with Block2 satisfied
   {@link BlockCacheUtil#shouldReplaceExistingCacheBlock}, so Block2 would
   replace Block1, but thread1 stopping before {@link BucketCache#cacheBlockWithWaitInternal}
3. Thread2 invoking {@link BucketCache#getBlock} with the same {@link BlockCacheKey},
   which returned Block1, the {@link RefCnt} of Block1 is 2.
4. Thread1 continues caching Block2, in {@link BucketCache.WriterThread#putIntoBackingMap},
   the old Block1 is freed directly which {@link RefCnt} is 2, but the Block1 is still used
   by Thread2 and the content of Block1 would be overwritten after it is freed, which may
   cause a serious error.
</pre>
hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCacheRefCnt.java