HBASE-27795: Define RPC API for cache cleaning

Author: shankot

hbase-client/src/main/java/org/apache/hadoop/hbase/client/Admin.java

@@ -2651,4 +2651,10 @@ List<LogEntry> getLogEntries(Set<ServerName> serverNames, String logType, Server
    * Get the list of cached files
    */
   List<String> getCachedFilesList(ServerName serverName) throws IOException;
+
+  /**
+   * Clean Cache by evicting the blocks of files belonging to regions
+   * that are no longer served by the RegionServer.
+   */
+  Map<String, Integer> cleanBucketCache(ServerName serverName) throws IOException;
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AdminOverAsyncAdmin.java

@@ -1136,4 +1136,12 @@ public void flushMasterStore() throws IOException {
   public List<String> getCachedFilesList(ServerName serverName) throws IOException {
     return get(admin.getCachedFilesList(serverName));
   }
+
+  /**
+   * Clean BucketCache
+   */
+  @Override
+  public Map<String,Integer> cleanBucketCache(ServerName serverName) throws IOException {
+    return get(admin.cleanBucketCache(serverName));
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java

@@ -1861,4 +1861,9 @@ CompletableFuture<List<LogEntry>> getLogEntries(Set<ServerName> serverNames, Str
    * Get the list of cached files
    */
   CompletableFuture<List<String>> getCachedFilesList(ServerName serverName);
+
+  /**
+   * Clean BucketCache
+   */
+  CompletableFuture<Map<String, Integer>> cleanBucketCache(ServerName serverName);
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java

@@ -1005,4 +1005,9 @@ public CompletableFuture<Void> flushMasterStore() {
   public CompletableFuture<List<String>> getCachedFilesList(ServerName serverName) {
     return wrap(rawAdmin.getCachedFilesList(serverName));
   }
+
+  @Override
+  public CompletableFuture<Map<String, Integer>> cleanBucketCache(ServerName serverName) {
+    return wrap(rawAdmin.cleanBucketCache(serverName));
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java

@@ -122,6 +122,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.RevokeRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.RevokeResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CleanBucketCacheRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CleanBucketCacheResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheRequest;
@@ -4537,4 +4539,18 @@ List<String>> adminCall(controller, stub, request.build(),
           resp -> resp.getCachedFilesList()))
       .serverName(serverName).call();
   }
+
+  /**
+   * Clean BucketCache
+   */
+  @Override
+  public CompletableFuture<Map<String, Integer>> cleanBucketCache(ServerName serverName) {
+    CleanBucketCacheRequest.Builder request = CleanBucketCacheRequest.newBuilder();
+    return this.<Map<String, Integer>> newAdminCaller()
+      .action((controller, stub) -> this.<CleanBucketCacheRequest, CleanBucketCacheResponse,
+        Map<String, Integer>> adminCall(controller, stub, request.build(),
+          (s, c, req, done) -> s.cleanBucketCache(c, req, done),
+          resp -> resp.getUncachedFilesMap()))
+      .serverName(serverName).call();
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java

@@ -152,6 +152,8 @@
 
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CleanBucketCacheRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CleanBucketCacheResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearSlowLogResponses;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CloseRegionRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetCachedFilesListRequest;
@@ -1835,6 +1837,21 @@ public static List<String> getCachedFilesList(final RpcController controller,
     return new ArrayList<>(response.getCachedFilesList());
   }
 
+  /**
+   * Get the list of cached files
+   */
+  public static Map<String, Integer> cleanBucketCache(final RpcController controller,
+    final AdminService.BlockingInterface admin) throws IOException {
+    CleanBucketCacheRequest request = CleanBucketCacheRequest.newBuilder().build();
+    CleanBucketCacheResponse response = null;
+    try {
+      response = admin.cleanBucketCache(controller, request);
+    } catch (ServiceException se) {
+      throw getRemoteException(se);
+    }
+    return response.getUncachedFilesMap();
+  }
+
   /**
    * Get the list of region info from a GetOnlineRegionResponse
    * @param proto the GetOnlineRegionResponse

hbase-protocol-shaded/src/main/protobuf/server/region/Admin.proto

@@ -336,6 +336,14 @@ message ClearSlowLogResponses {
   required bool is_cleaned = 1;
 }
 
+message CleanBucketCacheRequest {
+}
+
+message CleanBucketCacheResponse {
+  map<string, int32> uncached_files = 1;
+}
+
+
 service AdminService {
   rpc GetRegionInfo(GetRegionInfoRequest)
     returns(GetRegionInfoResponse);
@@ -415,4 +423,7 @@ service AdminService {
   rpc GetCachedFilesList(GetCachedFilesListRequest)
     returns(GetCachedFilesListResponse);
 
+  rpc CleanBucketCache(CleanBucketCacheRequest)
+    returns(CleanBucketCacheResponse);
+
 }

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

@@ -18,8 +18,10 @@
 package org.apache.hadoop.hbase.io.hfile;
 
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import org.apache.hadoop.hbase.regionserver.HStoreFile;
 import org.apache.yetus.audience.InterfaceAudience;
 
 /**
@@ -170,4 +172,18 @@ default boolean isMetaBlock(BlockType blockType) {
   default Optional<Map<String, Boolean>> getFullyCachedFiles() {
     return Optional.empty();
   }
+
+  /**
+   * Returns the list of files with stale blocks in cache.
+   */
+  default Optional<List<HStoreFile>> getFilesWithStaleBlocks() {
+    return Optional.empty();
+  }
+
+  /**
+   * Clean cache by evicting files with stale blocks in cache.
+   */
+  default Optional<Map<String, Integer>> cleanCacheByRemovingStaleBlocks(){
+    return Optional.empty();
+  }
 }

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

@@ -18,10 +18,12 @@
 package org.apache.hadoop.hbase.io.hfile;
 
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import org.apache.hadoop.hbase.io.HeapSize;
 import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
+import org.apache.hadoop.hbase.regionserver.HStoreFile;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -435,6 +437,16 @@ public Optional<Map<String, Boolean>> getFullyCachedFiles() {
     return this.l2Cache.getFullyCachedFiles();
   }
 
+  @Override
+  public Optional<List<HStoreFile>> getFilesWithStaleBlocks() {
+    return l2Cache.getFilesWithStaleBlocks();
+  }
+
+  @Override
+  public Optional<Map<String, Integer>> cleanCacheByRemovingStaleBlocks() {
+    return l2Cache.cleanCacheByRemovingStaleBlocks();
+  }
+
   @Override
   public void setMaxSize(long size) {
     this.l1Cache.setMaxSize(size);

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

@@ -26,6 +26,7 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -73,6 +74,9 @@
 import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.hbase.nio.RefCnt;
 import org.apache.hadoop.hbase.protobuf.ProtobufMagic;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HStore;
+import org.apache.hadoop.hbase.regionserver.HStoreFile;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.IdReadWriteLock;
@@ -156,6 +160,8 @@ public class BucketCache implements BlockCache, HeapSize {
 
   private BucketCachePersister cachePersister;
 
+  private static List<HStoreFile> filesWithStaleBlocks = new ArrayList<>();
+
   /**
    * Flag if the cache is enabled or not... We shut it off if there are IO errors for some time, so
    * that Bucket IO exceptions/errors don't bring down the HBase server.
@@ -1985,4 +1991,28 @@ public static Optional<BucketCache> getBucketCacheFromCacheConfig(CacheConfig ca
     return Optional.empty();
   }
 
+  @Override
+  public Optional<List<HStoreFile>> getFilesWithStaleBlocks() {
+    return Optional.of(filesWithStaleBlocks);
+  }
+
+  public static void clearFilesWithStaleBlocks() {
+    filesWithStaleBlocks.clear();
+  }
+
+  public void setFilesWithStaleBlocks(HRegion hRegion) {
+    for (HStore hs : hRegion.getStores()) {
+      filesWithStaleBlocks.addAll(hs.getStorefiles());
+    }
+  }
+
+  @Override
+  public Optional<Map<String, Integer>> cleanCacheByRemovingStaleBlocks() {
+    Map<String, Integer> evictedFilesWithStaleBlocks = new HashMap<>();
+    for (HStoreFile hsf : filesWithStaleBlocks) {
+      evictedFilesWithStaleBlocks.put(hsf.getPath().getName(), evictBlocksByHfileName(hsf.getPath().getName()));
+    }
+    return Optional.of(evictedFilesWithStaleBlocks);
+  }
+
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java

@@ -150,6 +150,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.RevokeRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.RevokeResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CleanBucketCacheRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CleanBucketCacheResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheRequest;
@@ -3399,6 +3401,12 @@ public HBaseProtos.LogEntry getLogEntries(RpcController controller,
     throw new ServiceException("Invalid request params");
   }
 
+  @Override
+  public CleanBucketCacheResponse cleanBucketCache(RpcController controller,
+    CleanBucketCacheRequest request) throws ServiceException {
+    throw new ServiceException(new DoNotRetryIOException("Unsupported method on master"));
+  }
+
   private MasterProtos.BalancerDecisionsResponse
     getBalancerDecisions(MasterProtos.BalancerDecisionsRequest request) {
     final NamedQueueRecorder namedQueueRecorder = this.server.getNamedQueueRecorder();

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -1652,6 +1652,7 @@ public Map<byte[], List<HStoreFile>> close(boolean abort, boolean ignoreStatus,
                       + "so setting evict on close to false. ",
                     this.getRegionInfo().getRegionNameAsString());
                   this.getStores().forEach(s -> s.getCacheConfig().setEvictOnClose(false));
+                  ((BucketCache) l2).setFilesWithStaleBlocks(this);
                 }
               }
             }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java

@@ -87,6 +87,7 @@
 import org.apache.hadoop.hbase.exceptions.UnknownProtocolException;
 import org.apache.hadoop.hbase.io.ByteBuffAllocator;
 import org.apache.hadoop.hbase.io.hfile.BlockCache;
+import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
 import org.apache.hadoop.hbase.ipc.PriorityFunction;
 import org.apache.hadoop.hbase.ipc.QosPriority;
@@ -155,6 +156,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
 import org.apache.hadoop.hbase.shaded.protobuf.ResponseConverter;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CleanBucketCacheRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CleanBucketCacheResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheRequest;
@@ -3946,4 +3949,16 @@ public GetCachedFilesListResponse getCachedFilesList(RpcController controller,
     });
     return responseBuilder.addAllCachedFiles(fullyCachedFiles).build();
   }
+
+  @Override
+  public CleanBucketCacheResponse cleanBucketCache(RpcController controller,
+    CleanBucketCacheRequest request) throws ServiceException {
+    CleanBucketCacheResponse.Builder responseBuilder = CleanBucketCacheResponse.newBuilder();
+    Map<String, Integer> evictedFilesWithStaleBlocks = new HashMap<>();
+    server.getBlockCache().flatMap(BlockCache::cleanCacheByRemovingStaleBlocks)
+      .ifPresent(evictedFilesWithStaleBlocks::putAll);
+    responseBuilder.putAllUncachedFiles(evictedFilesWithStaleBlocks);
+    BucketCache.clearFilesWithStaleBlocks();
+    return responseBuilder.build();
+  }
 }