Backport HBASE-28804 to branch-2

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

@@ -246,4 +246,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

@@ -497,4 +497,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;
                   }