HBASE-28804: Implement asynchronous retrieval of bucket-cache data from persistence

hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/balancer/CacheAwareLoadBalancer.java

@@ -28,6 +28,7 @@
 
 import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_PERSISTENT_PATH_KEY;
 
+import java.text.DecimalFormat;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -221,6 +222,9 @@ private boolean moveRegionToOldServer(BalancerClusterState cluster, int regionIn
         return false;
       }
 
+      DecimalFormat df = new DecimalFormat("#");
+      df.setMaximumFractionDigits(4);
+
       float cacheRatioDiffThreshold = 0.6f;
 
       // Conditions for moving the region
@@ -240,7 +244,7 @@ private boolean moveRegionToOldServer(BalancerClusterState cluster, int regionIn
           LOG.debug(
             "Region {} moved from {} to {} as the region is cached {} equally on both servers",
             cluster.regions[regionIndex].getEncodedName(), cluster.servers[currentServerIndex],
-            cluster.servers[oldServerIndex], cacheRatioOnCurrentServer);
+            cluster.servers[oldServerIndex], df.format(cacheRatioOnCurrentServer));
         }
         return true;
       }
@@ -257,7 +261,8 @@ private boolean moveRegionToOldServer(BalancerClusterState cluster, int regionIn
             "Region {} moved from {} to {} as region cache ratio {} is better than the current "
               + "cache ratio {}",
             cluster.regions[regionIndex].getEncodedName(), cluster.servers[currentServerIndex],
-            cluster.servers[oldServerIndex], cacheRatioOnCurrentServer, cacheRatioOnOldServer);
+            cluster.servers[oldServerIndex], cacheRatioOnCurrentServer,
+            df.format(cacheRatioOnCurrentServer));
         }
         return true;
       }
@@ -266,7 +271,8 @@ private boolean moveRegionToOldServer(BalancerClusterState cluster, int regionIn
         LOG.debug(
           "Region {} not moved from {} to {} with current cache ratio {} and old cache ratio {}",
           cluster.regions[regionIndex], cluster.servers[currentServerIndex],
-          cluster.servers[oldServerIndex], cacheRatioOnCurrentServer, cacheRatioOnOldServer);
+          cluster.servers[oldServerIndex], cacheRatioOnCurrentServer,
+          df.format(cacheRatioOnCurrentServer));
       }
       return false;
     }

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

@@ -261,4 +261,21 @@ default Optional<Map<String, Long>> getRegionCachedInfo() {
   default int evictBlocksRangeByHfileName(String hfileName, long initOffset, long endOffset) {
     return 0;
   }
+
+  /**
+   * API to check whether or not, the cache is enabled.
+   * @return returns true if the cache is enabled, false otherwise.
+   */
+  default boolean isCacheEnabled() {
+    return true;
+  }
+
+  /**
+   * Wait for the bucket cache to be enabled while server restart
+   * @param timeout time to wait for the bucket cache to be enable
+   * @return boolean true if the bucket cache is enabled, false otherwise
+   */
+  default boolean waitForCacheInitialization(long timeout) {
+    return true;
+  }
 }

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

@@ -507,4 +507,15 @@ public int evictBlocksRangeByHfileName(String hfileName, long initOffset, long e
     return l1Cache.evictBlocksRangeByHfileName(hfileName, initOffset, endOffset)
       + l2Cache.evictBlocksRangeByHfileName(hfileName, initOffset, endOffset);
   }
+
+  @Override
+  public boolean waitForCacheInitialization(long timeout) {
+    return this.l1Cache.waitForCacheInitialization(timeout)
+      && this.l2Cache.waitForCacheInitialization(timeout);
+  }
+
+  @Override
+  public boolean isCacheEnabled() {
+    return l1Cache.isCacheEnabled() && l2Cache.isCacheEnabled();
+  }
 }

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

@@ -34,12 +34,15 @@
 public class HFilePreadReader extends HFileReaderImpl {
   private static final Logger LOG = LoggerFactory.getLogger(HFileReaderImpl.class);
 
+  private static final int WAIT_TIME_FOR_CACHE_INITIALIZATION = 10 * 60 * 1000;
+
   public HFilePreadReader(ReaderContext context, HFileInfo fileInfo, CacheConfig cacheConf,
     Configuration conf) throws IOException {
     super(context, fileInfo, cacheConf, conf);
     final MutableBoolean shouldCache = new MutableBoolean(true);
 
     cacheConf.getBlockCache().ifPresent(cache -> {
+      cache.waitForCacheInitialization(WAIT_TIME_FOR_CACHE_INITIALIZATION);
       Optional<Boolean> result = cache.shouldCacheFile(path.getName());
       shouldCache.setValue(result.isPresent() ? result.get().booleanValue() : true);
     });
@@ -110,8 +113,8 @@ public void run() {
                   if (!cache.blockFitsIntoTheCache(block).orElse(true)) {
                     LOG.warn(
                       "Interrupting prefetch for file {} because block {} of size {} "
-                        + "doesn't fit in the available cache space.",
-                      path, cacheKey, block.getOnDiskSizeWithHeader());
+                        + "doesn't fit in the available cache space. isCacheEnabled: {}",
+                      path, cacheKey, block.getOnDiskSizeWithHeader(), cache.isCacheEnabled());
                     interrupted = true;
                     break;
                   }