HBASE-27795: Define RPC API for cache cleaning

shankot · shankot · commit 6a1fd445db63 · 2023-12-01T12:34:27.000-06:00
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Admin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Admin.java
@@ -2651,4 +2651,13 @@ 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.
+   * @param serverName ServerName
+   * @return A map of filename and number of blocks evicted.
+   * @throws IOException if a remote or network exception occurs
+   */
+  Map<String, Integer> uncacheStaleBlocks(ServerName serverName) throws IOException;
 }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AdminOverAsyncAdmin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AdminOverAsyncAdmin.java
@@ -1136,4 +1136,9 @@ public void flushMasterStore() throws IOException {
   public List<String> getCachedFilesList(ServerName serverName) throws IOException {
     return get(admin.getCachedFilesList(serverName));
   }
+
+  @Override
+  public Map<String, Integer> uncacheStaleBlocks(ServerName serverName) throws IOException {
+    return get(admin.uncacheStaleBlocks(serverName));
+  }
 }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
@@ -1861,4 +1861,12 @@ CompletableFuture<List<LogEntry>> getLogEntries(Set<ServerName> serverNames, Str
    * Get the list of cached files
    */
   CompletableFuture<List<String>> getCachedFilesList(ServerName serverName);
+
+  /**
+   * Clean Cache by evicting the blocks of files belonging to regions that are no longer served by
+   * the RegionServer.
+   * @param serverName ServerName
+   * @return A map of filename and number of blocks evicted.
+   */
+  CompletableFuture<Map<String, Integer>> uncacheStaleBlocks(ServerName serverName);
 }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java b/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>> uncacheStaleBlocks(ServerName serverName) {
+    return wrap(rawAdmin.uncacheStaleBlocks(serverName));
+  }
 }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java
@@ -144,6 +144,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RollWALWriterResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UncacheStaleBlocksRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UncacheStaleBlocksResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
@@ -4537,4 +4539,15 @@ List<String>> adminCall(controller, stub, request.build(),
           resp -> resp.getCachedFilesList()))
       .serverName(serverName).call();
   }
+
+  @Override
+  public CompletableFuture<Map<String, Integer>> uncacheStaleBlocks(ServerName serverName) {
+    UncacheStaleBlocksRequest.Builder request = UncacheStaleBlocksRequest.newBuilder();
+    return this.<Map<String, Integer>> newAdminCaller()
+      .action((controller, stub) -> this.<UncacheStaleBlocksRequest, UncacheStaleBlocksResponse,
+        Map<String, Integer>> adminCall(controller, stub, request.build(),
+          (s, c, req, done) -> s.uncacheStaleBlocks(c, req, done),
+          resp -> resp.getUncachedFilesMap()))
+      .serverName(serverName).call();
+  }
 }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
@@ -166,6 +166,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetStoreFileResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.OpenRegionRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ServerInfo;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UncacheStaleBlocksRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UncacheStaleBlocksResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WarmupRegionRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.CellProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
@@ -1835,6 +1837,22 @@ public static List<String> getCachedFilesList(final RpcController controller,
     return new ArrayList<>(response.getCachedFilesList());
   }
 
+  /**
+   * Clean Cache by evicting the blocks of files belonging to regions that are no longer served by
+   * the RegionServer.
+   */
+  public static Map<String, Integer> uncacheStaleBlocks(final RpcController controller,
+    final AdminService.BlockingInterface admin) throws IOException {
+    UncacheStaleBlocksRequest request = UncacheStaleBlocksRequest.newBuilder().build();
+    UncacheStaleBlocksResponse response = null;
+    try {
+      response = admin.uncacheStaleBlocks(controller, request);
+    } catch (ServiceException se) {
+      throw getRemoteException(se);
+    }
+    return response.getUncachedFilesMap();
+  }
+
   /**
    * Get the list of region info from a GetOnlineRegionResponse
    * @param proto the GetOnlineRegionResponse
diff --git a/hbase-protocol-shaded/src/main/protobuf/server/region/Admin.proto b/hbase-protocol-shaded/src/main/protobuf/server/region/Admin.proto
@@ -336,6 +336,14 @@ message ClearSlowLogResponses {
   required bool is_cleaned = 1;
 }
 
+message UncacheStaleBlocksRequest {
+}
+
+message UncacheStaleBlocksResponse {
+  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 UncacheStaleBlocks(UncacheStaleBlocksRequest)
+    returns(UncacheStaleBlocksResponse);
+
 }
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockCache.java
@@ -21,6 +21,7 @@
 import java.util.Map;
 import java.util.Optional;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.regionserver.RegionAvailabilityChecker;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -245,4 +246,15 @@ default Optional<Integer> getBlockSize(BlockCacheKey key) {
   default Optional<Map<String, Pair<String, Long>>> getFullyCachedFiles() {
     return Optional.empty();
   }
+
+  /**
+   * Clean Cache by evicting the blocks of files belonging to regions that are no longer served by
+   * the RegionServer.
+   * @param regionAvailabilityChecker RegionAvailabilityChecker
+   * @return A map of filename and number of blocks evicted.
+   */
+  default Optional<Map<String, Integer>>
+    uncacheStaleBlocks(RegionAvailabilityChecker regionAvailabilityChecker) {
+    return Optional.empty();
+  }
 }
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CombinedBlockCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CombinedBlockCache.java
@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase.io.hfile;
 
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Optional;
@@ -25,6 +26,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.io.HeapSize;
 import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
+import org.apache.hadoop.hbase.regionserver.RegionAvailabilityChecker;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -439,6 +441,16 @@ public Optional<Map<String, Pair<String, Long>>> getFullyCachedFiles() {
     return this.l2Cache.getFullyCachedFiles();
   }
 
+  @Override
+  public Optional<Map<String, Integer>>
+    uncacheStaleBlocks(RegionAvailabilityChecker regionAvailabilityChecker) {
+    Map<String, Integer> uncachedStaleBlocksMap =
+      l1Cache.uncacheStaleBlocks(regionAvailabilityChecker).orElseGet(HashMap::new);
+    l2Cache.uncacheStaleBlocks(regionAvailabilityChecker).ifPresent(
+      map2 -> map2.forEach((key, value) -> uncachedStaleBlocksMap.merge(key, value, Integer::sum)));
+    return Optional.of(uncachedStaleBlocksMap);
+  }
+
   @Override
   public void setMaxSize(long size) {
     this.l1Cache.setMaxSize(size);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
@@ -27,6 +27,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -76,6 +77,7 @@
 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.RegionAvailabilityChecker;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.IdReadWriteLock;
@@ -2144,4 +2146,21 @@ public Optional<Integer> getBlockSize(BlockCacheKey key) {
     }
 
   }
+
+  public Optional<Map<String, Integer>>
+    uncacheStaleBlocks(RegionAvailabilityChecker regionAvailabilityChecker) {
+    Map<String, Integer> evictedFilesWithStaleBlocks = new HashMap<>();
+
+    fullyCachedFiles.forEach((fileName, value) -> {
+      int blocksEvicted = (!regionAvailabilityChecker.isRegionServedByTheServer(value.getFirst()))
+        ? evictBlocksByHfileName(fileName)
+        : 0;
+      evictedFilesWithStaleBlocks.put(fileName, blocksEvicted);
+      LOG.debug(
+        "Uncached {} blocks belonging to the file {} as the region {} "
+          + "is not served by the region server anymore.",
+        blocksEvicted, fileName, value.getFirst());
+    });
+    return Optional.of(evictedFilesWithStaleBlocks);
+  }
 }
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java
@@ -184,6 +184,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RollWALWriterResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UncacheStaleBlocksRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UncacheStaleBlocksResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateFavoredNodesRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateFavoredNodesResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WarmupRegionRequest;
@@ -3399,6 +3401,12 @@ public HBaseProtos.LogEntry getLogEntries(RpcController controller,
     throw new ServiceException("Invalid request params");
   }
 
+  @Override
+  public UncacheStaleBlocksResponse uncacheStaleBlocks(RpcController controller,
+    UncacheStaleBlocksRequest 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();
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
@@ -494,6 +494,11 @@ public class HRegionServer extends HBaseServerBase<RSRpcServices>
    */
   private ReplicationMarkerChore replicationMarkerChore;
 
+  final RegionAvailabilityChecker regionAvailabilityChecker = encodedRegionName -> {
+    HRegion regionOnServer = getRegion(encodedRegionName);
+    return (regionOnServer != null && regionOnServer.isAvailable());
+  };
+
   /**
    * Starts a HRegionServer at the default location.
    * <p/>
@@ -2692,6 +2697,10 @@ public RegionStoreSequenceIds getLastSequenceId(byte[] encodedRegionName) {
     }
   }
 
+  RegionAvailabilityChecker getRegionAvailabilityChecker() {
+    return this.regionAvailabilityChecker;
+  }
+
   /**
    * Close meta region if we carry it
    * @param abort Whether we're running an abort.
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
@@ -192,6 +192,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RollWALWriterResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UncacheStaleBlocksRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UncacheStaleBlocksResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateFavoredNodesRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateFavoredNodesResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WALEntry;
@@ -3946,4 +3948,16 @@ public GetCachedFilesListResponse getCachedFilesList(RpcController controller,
     });
     return responseBuilder.addAllCachedFiles(fullyCachedFiles).build();
   }
+
+  @Override
+  public UncacheStaleBlocksResponse uncacheStaleBlocks(RpcController controller,
+    UncacheStaleBlocksRequest request) throws ServiceException {
+    UncacheStaleBlocksResponse.Builder responseBuilder = UncacheStaleBlocksResponse.newBuilder();
+    Map<String, Integer> evictedFilesWithStaleBlocks = new HashMap<>();
+    server.getBlockCache()
+      .flatMap(bc -> bc.uncacheStaleBlocks(server.getRegionAvailabilityChecker()))
+      .ifPresent(evictedFilesWithStaleBlocks::putAll);
+    responseBuilder.putAllUncachedFiles(evictedFilesWithStaleBlocks);
+    return responseBuilder.build();
+  }
 }
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionAvailabilityChecker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionAvailabilityChecker.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.regionserver;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+@FunctionalInterface
+public interface RegionAvailabilityChecker {
+
+  /**
+   * Checks whether the region is served by the RegionServer. Intended for internal use, this method
+   * avoids passing the entire HRegionServer object when checking the availability of the region on
+   * the region server.
+   * @param encodedRegionName The encoded region name as a String.
+   * @return True if the region is present and available on the region server; otherwise, false.
+   */
+  boolean isRegionServedByTheServer(String encodedRegionName);
+}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCleanBucketCache.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCleanBucketCache.java
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/MockRegionServer.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/MockRegionServer.java
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/rsgroup/VerifyingRSGroupAdmin.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/rsgroup/VerifyingRSGroupAdmin.java
diff --git a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/client/ThriftAdmin.java b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/client/ThriftAdmin.java

Original file line number	Diff line number	Diff line change
`@@ -1136,4 +1136,9 @@ public void flushMasterStore() throws IOException {`
`1136`	`1136`	`public List<String> getCachedFilesList(ServerName serverName) throws IOException {`
`1137`	`1137`	`return get(admin.getCachedFilesList(serverName));`
`1138`	`1138`	`}`
	`1139`	`+`
	`1140`	`+ @Override`
	`1141`	`+ public Map<String, Integer> uncacheStaleBlocks(ServerName serverName) throws IOException {`
	`1142`	`+ return get(admin.uncacheStaleBlocks(serverName));`
	`1143`	`+ }`
`1139`	`1144`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1005,4 +1005,9 @@ public CompletableFuture<Void> flushMasterStore() {`
`1005`	`1005`	`public CompletableFuture<List<String>> getCachedFilesList(ServerName serverName) {`
`1006`	`1006`	`return wrap(rawAdmin.getCachedFilesList(serverName));`
`1007`	`1007`	`}`
	`1008`	`+`
	`1009`	`+ @Override`
	`1010`	`+ public CompletableFuture<Map<String, Integer>> uncacheStaleBlocks(ServerName serverName) {`
	`1011`	`+ return wrap(rawAdmin.uncacheStaleBlocks(serverName));`
	`1012`	`+ }`
`1008`	`1013`	`}`