HBASE-27750: Update the list of prefetched Hfiles upon block eviction (#5140)

Signed-off-by: Wellington Chevreuil <wchevreuil@apache.org>

Author: Kota-SH

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

@@ -379,6 +379,7 @@ protected void startWriterThreads() {
   void startBucketCachePersisterThread() {
     BucketCachePersister cachePersister =
       new BucketCachePersister(this, bucketcachePersistInterval);
+    cachePersister.setDaemon(true);
     cachePersister.start();
   }
 
@@ -610,6 +611,9 @@ void blockEvicted(BlockCacheKey cacheKey, BucketEntry bucketEntry, boolean decre
       cacheStats.evicted(bucketEntry.getCachedTime(), cacheKey.isPrimary());
     }
     if (ioEngine.isPersistent()) {
+      if (prefetchedFileListPath != null) {
+        PrefetchExecutor.removePrefetchedFileWhileEvict(cacheKey.getHfileName());
+      }
       setCacheInconsistent(true);
     }
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestPrefetchRSClose.java

@@ -108,7 +108,7 @@ public void testPrefetchPersistence() throws Exception {
       table.put(put1);
       TEST_UTIL.flush(tableName);
     } finally {
-      Thread.sleep(1500);
+      Thread.sleep(2000);
     }
 
     // Default interval for cache persistence is 1000ms. So after 1000ms, both the persistence files

hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCachePersister.java

@@ -38,6 +38,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.PrefetchExecutor;
 import org.apache.hadoop.hbase.io.hfile.RandomKeyValueUtil;
 import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
 import org.apache.hadoop.hbase.testclassification.IOTests;
@@ -119,21 +120,36 @@ public void testPrefetchPersistenceCrash() throws Exception {
 
   @Test
   public void testPrefetchPersistenceCrashNegative() throws Exception {
-    long bucketCachePersistInterval = 3000;
+    long bucketCachePersistInterval = Long.MAX_VALUE;
     Configuration conf = setupBucketCacheConfig(bucketCachePersistInterval);
     BucketCache bucketCache = setupBucketCache(conf);
     CacheConfig cacheConf = new CacheConfig(conf, bucketCache);
     FileSystem fs = HFileSystem.get(conf);
     // Load Cache
     Path storeFile = writeStoreFile("TestPrefetch2", conf, cacheConf, fs);
-    Path storeFile2 = writeStoreFile("TestPrefetch3", conf, cacheConf, fs);
     readStoreFile(storeFile, 0, fs, cacheConf, conf, bucketCache);
-    readStoreFile(storeFile2, 0, fs, cacheConf, conf, bucketCache);
     assertFalse(new File(testDir + "/prefetch.persistence").exists());
     assertFalse(new File(testDir + "/bucket.persistence").exists());
     cleanupBucketCache(bucketCache);
   }
 
+  @Test
+  public void testPrefetchListUponBlockEviction() throws Exception {
+    Configuration conf = setupBucketCacheConfig(200);
+    BucketCache bucketCache1 = setupBucketCache(conf);
+    CacheConfig cacheConf = new CacheConfig(conf, bucketCache1);
+    FileSystem fs = HFileSystem.get(conf);
+    // Load Blocks in cache
+    Path storeFile = writeStoreFile("TestPrefetch3", conf, cacheConf, fs);
+    readStoreFile(storeFile, 0, fs, cacheConf, conf, bucketCache1);
+    Thread.sleep(500);
+    // Evict Blocks from cache
+    BlockCacheKey bucketCacheKey = bucketCache1.backingMap.entrySet().iterator().next().getKey();
+    assertTrue(PrefetchExecutor.isFilePrefetched(storeFile.getName()));
+    bucketCache1.evictBlock(bucketCacheKey);
+    assertFalse(PrefetchExecutor.isFilePrefetched(storeFile.getName()));
+  }
+
   public void readStoreFile(Path storeFilePath, long offset, FileSystem fs, CacheConfig cacheConf,
     Configuration conf, BucketCache bucketCache) throws Exception {
     // Open the file