HBASE-28467: Addressed further review comments.

Change-Id: Ia55e2e6ee415fa68d1c2714d127670f6df9c404d

Author: jhungund

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

@@ -275,16 +275,13 @@ public boolean shouldCacheBlockOnRead(BlockCategory category) {
       || (prefetchOnOpen && (category != BlockCategory.META && category != BlockCategory.UNKNOWN));
   }
 
-  public boolean shouldCacheFileBlock(HFileInfo hFileInfo, Configuration conf) {
+  public boolean shouldCacheBlockOnRead(BlockCategory category, HFileInfo hFileInfo,
+    Configuration conf) {
     Optional<Boolean> cacheFileBlock = Optional.of(true);
-    // Additionally perform the time-based priority checks to see
-    // whether, or not to cache the block.
     if (getBlockCache().isPresent()) {
-
       cacheFileBlock = getBlockCache().get().shouldCacheFile(hFileInfo, conf);
-      LOG.info("BlockCache Present, cacheFileBlock: {}", cacheFileBlock.get());
     }
-    return cacheFileBlock.get();
+    return shouldCacheBlockOnRead(category) && cacheFileBlock.get();
   }
 
   /** Returns true if blocks in this file should be flagged as in-memory */

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

@@ -1193,7 +1193,8 @@ public HFileBlock getMetaBlock(String metaBlockName, boolean cacheBlock) throws
       BlockCacheKey cacheKey =
         new BlockCacheKey(name, metaBlockOffset, this.isPrimaryReplicaReader(), BlockType.META);
 
-      cacheBlock &= cacheConf.shouldCacheBlockOnRead(BlockType.META.getCategory());
+      cacheBlock &=
+        cacheConf.shouldCacheBlockOnRead(BlockType.META.getCategory(), getHFileInfo(), conf);
       HFileBlock cachedBlock =
         getCachedBlock(cacheKey, cacheBlock, false, true, BlockType.META, null);
       if (cachedBlock != null) {
@@ -1346,15 +1347,15 @@ public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize, final bo
         }
         BlockType.BlockCategory category = hfileBlock.getBlockType().getCategory();
         final boolean cacheCompressed = cacheConf.shouldCacheCompressed(category);
-        final boolean cacheOnRead = cacheConf.shouldCacheBlockOnRead(category);
-        final boolean shouldCacheFileBlock = cacheConf.shouldCacheFileBlock(getHFileInfo(), conf);
+        final boolean cacheOnRead =
+          cacheConf.shouldCacheBlockOnRead(category, getHFileInfo(), conf);
 
         // Don't need the unpacked block back and we're storing the block in the cache compressed
-        if (cacheOnly && cacheCompressed && cacheOnRead && shouldCacheFileBlock) {
+        if (cacheOnly && cacheCompressed && cacheOnRead) {
           cacheConf.getBlockCache().ifPresent(cache -> {
             LOG.debug("Skipping decompression of block {} in prefetch", cacheKey);
             // Cache the block if necessary
-            if (cacheable && cacheConf.shouldCacheBlockOnRead(category)) {
+            if (cacheable && cacheOnRead) {
               cache.cacheBlock(cacheKey, hfileBlock, cacheConf.isInMemory(), cacheOnly);
             }
           });
@@ -1367,7 +1368,7 @@ public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize, final bo
         HFileBlock unpacked = hfileBlock.unpack(hfileContext, fsBlockReader);
         // Cache the block if necessary
         cacheConf.getBlockCache().ifPresent(cache -> {
-          if (cacheable && cacheConf.shouldCacheBlockOnRead(category) && shouldCacheFileBlock) {
+          if (cacheable && cacheOnRead) {
             // Using the wait on cache during compaction and prefetching.
             cache.cacheBlock(cacheKey, cacheCompressed ? hfileBlock : unpacked,
               cacheConf.isInMemory(), cacheOnly);

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

@@ -55,6 +55,7 @@
 import org.apache.hadoop.hbase.io.hfile.BlockCacheFactory;
 import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
 import org.apache.hadoop.hbase.io.hfile.BlockType;
+import org.apache.hadoop.hbase.io.hfile.BlockType.BlockCategory;
 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
 import org.apache.hadoop.hbase.io.hfile.CacheTestUtils;
 import org.apache.hadoop.hbase.io.hfile.HFileBlock;
@@ -484,13 +485,17 @@ public void testCacheConfigShouldCacheFile() throws Exception {
     // hStoreFiles[0], hStoreFiles[1], hStoreFiles[2] are hot files.
     // hStoreFiles[3] is a cold file.
     try {
-      assertTrue(cacheConf.shouldCacheFileBlock(hStoreFiles.get(0).getFileInfo().getHFileInfo(),
+      assertTrue(cacheConf.shouldCacheBlockOnRead(BlockCategory.DATA,
+        hStoreFiles.get(0).getFileInfo().getHFileInfo(),
         hStoreFiles.get(0).getFileInfo().getConf()));
-      assertTrue(cacheConf.shouldCacheFileBlock(hStoreFiles.get(1).getFileInfo().getHFileInfo(),
+      assertTrue(cacheConf.shouldCacheBlockOnRead(BlockCategory.DATA,
+        hStoreFiles.get(1).getFileInfo().getHFileInfo(),
         hStoreFiles.get(1).getFileInfo().getConf()));
-      assertTrue(cacheConf.shouldCacheFileBlock(hStoreFiles.get(2).getFileInfo().getHFileInfo(),
+      assertTrue(cacheConf.shouldCacheBlockOnRead(BlockCategory.DATA,
+        hStoreFiles.get(2).getFileInfo().getHFileInfo(),
         hStoreFiles.get(2).getFileInfo().getConf()));
-      assertFalse(cacheConf.shouldCacheFileBlock(hStoreFiles.get(3).getFileInfo().getHFileInfo(),
+      assertFalse(cacheConf.shouldCacheBlockOnRead(BlockCategory.DATA,
+        hStoreFiles.get(3).getFileInfo().getHFileInfo(),
         hStoreFiles.get(3).getFileInfo().getConf()));
     } finally {
       for (HStoreFile file : hStoreFiles) {
@@ -509,7 +514,7 @@ public void testCacheOnReadColdFile() throws Exception {
 
   @Test
   public void testCacheOnReadHotFile() throws Exception {
-    // hStoreFiles[0] is a hot file. the blocks should not get loaded after a readBlock call.
+    // hStoreFiles[0] is a hot file. the blocks should get loaded after a readBlock call.
     HStoreFile hStoreFile = hStoreFiles.get(0);
     BlockCacheKey cacheKey =
       new BlockCacheKey(hStoreFiles.get(0).getPath(), 0, true, BlockType.DATA);