HBASE-28146: Make ServerManager rsAdmins map thread safe (#5461)

Co-authored-by: Ray Mattingly <rmattingly@hubspot.com>
Signed-off-by: Duo Zhang <zhangduo@apache.org>
Signed-off-by: Bryan Beaudreault <bbeaudreault@apache.org>
(cherry picked from commit 1641a4a)

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

@@ -23,7 +23,6 @@
 import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -124,12 +123,6 @@ public class ServerManager {
   private final ConcurrentNavigableMap<ServerName, ServerMetrics> onlineServers =
     new ConcurrentSkipListMap<>();
 
-  /**
-   * Map of admin interfaces per registered regionserver; these interfaces we use to control
-   * regionservers out on the cluster
-   */
-  private final Map<ServerName, AdminService.BlockingInterface> rsAdmins = new HashMap<>();
-
   /** List of region servers that should not get any more new regions. */
   private final ArrayList<ServerName> drainingServers = new ArrayList<>();
 
@@ -397,7 +390,6 @@ private ServerName findServerWithSameHostnamePortWithLock(final ServerName serve
   void recordNewServerWithLock(final ServerName serverName, final ServerMetrics sl) {
     LOG.info("Registering regionserver=" + serverName);
     this.onlineServers.put(serverName, sl);
-    this.rsAdmins.remove(serverName);
   }
 
   public RegionStoreSequenceIds getLastFlushedSequenceId(byte[] encodedRegionName) {
@@ -599,7 +591,6 @@ public synchronized void moveFromOnlineToDeadServers(final ServerName sn) {
         LOG.trace("Expiration of {} but server not online", sn);
       }
     }
-    this.rsAdmins.remove(sn);
   }
 
   /*
@@ -711,18 +702,13 @@ public static void closeRegionSilentlyAndWait(ClusterConnection connection, Serv
    * @throws RetriesExhaustedException wrapping a ConnectException if failed
    */
   public AdminService.BlockingInterface getRsAdmin(final ServerName sn) throws IOException {
-    AdminService.BlockingInterface admin = this.rsAdmins.get(sn);
-    if (admin == null) {
-      LOG.debug("New admin connection to " + sn.toString());
-      if (sn.equals(master.getServerName()) && master instanceof HRegionServer) {
-        // A master is also a region server now, see HBASE-10569 for details
-        admin = ((HRegionServer) master).getRSRpcServices();
-      } else {
-        admin = this.connection.getAdmin(sn);
-      }
-      this.rsAdmins.put(sn, admin);
+    LOG.debug("New admin connection to {}", sn);
+    if (sn.equals(master.getServerName()) && master instanceof HRegionServer) {
+      // A master is also a region server now, see HBASE-10569 for details
+      return ((HRegionServer) master).getRSRpcServices();
+    } else {
+      return this.connection.getAdmin(sn);
     }
-    return admin;
   }
 
   /**