HBASE-25292 Improve InetSocketAddress usage discipline (#2669)

Network identities should be bound late. Remote addresses should be
resolved at the last possible moment, just before connect(). Network
identity mappings can change, so our code should not inappropriately
cache them. Otherwise we might miss a change and fail to operate normally.

Revert "HBASE-14544 Allow HConnectionImpl to not refresh the dns on errors"
Removes hbase.resolve.hostnames.on.failure and related code. We always
resolve hostnames, as late as possible.

Preserve InetSocketAddress caching per RPC connection. Avoids potential
lookups per Call.

Replace InetSocketAddress with Address where used as a map key. If we want
to key by hostname and/or resolved address we should be explicit about it.
Using Address chooses mapping by hostname and port only.

Add metrics for potential nameservice resolution attempts, whenever an
InetSocketAddress is instantiated for connect; and metrics for failed
resolution, whenever InetSocketAddress#isUnresolved on the new instance
is true.

* Use ServerName directly to build a stub key

* Resolve and cache ISA on a RpcChannel as late as possible, at first call

* Remove now invalid unit test TestCIBadHostname

We resolve DNS at the latest possible time, at first call, and do not
resolve hostnames for creating stubs at all, so this unit test cannot
work now.

Reviewed-by: Mingliang Liu <liuml07@apache.org>
Signed-off-by: Duo Zhang <zhangduo@apache.org>

Author: apurtell

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

@@ -28,6 +28,8 @@
 import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
 
 import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -74,8 +76,6 @@ class AsyncConnectionImpl implements AsyncConnection {
       .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(),
     10, TimeUnit.MILLISECONDS);
 
-  private static final String RESOLVE_HOSTNAME_ON_FAIL_KEY = "hbase.resolve.hostnames.on.failure";
-
   private final Configuration conf;
 
   final AsyncConnectionConfiguration connConf;
@@ -90,8 +90,6 @@ class AsyncConnectionImpl implements AsyncConnection {
 
   final RpcControllerFactory rpcControllerFactory;
 
-  private final boolean hostnameCanChange;
-
   private final AsyncRegionLocator locator;
 
   final AsyncRpcRetryingCallerFactory callerFactory;
@@ -134,7 +132,6 @@ public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, Stri
     }
     this.rpcClient = RpcClientFactory.createClient(conf, clusterId, metrics.orElse(null));
     this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
-    this.hostnameCanChange = conf.getBoolean(RESOLVE_HOSTNAME_ON_FAIL_KEY, true);
     this.rpcTimeout =
       (int) Math.min(Integer.MAX_VALUE, TimeUnit.NANOSECONDS.toMillis(connConf.getRpcTimeoutNs()));
     this.locator = new AsyncRegionLocator(this, RETRY_TIMER);
@@ -250,7 +247,7 @@ private ClientService.Interface createRegionServerStub(ServerName serverName) th
 
   ClientService.Interface getRegionServerStub(ServerName serverName) throws IOException {
     return ConcurrentMapUtils.computeIfAbsentEx(rsStubs,
-      getStubKey(ClientService.getDescriptor().getName(), serverName, hostnameCanChange),
+      getStubKey(ClientService.getDescriptor().getName(), serverName),
       () -> createRegionServerStub(serverName));
   }
 
@@ -264,7 +261,7 @@ private AdminService.Interface createAdminServerStub(ServerName serverName) thro
 
   AdminService.Interface getAdminStub(ServerName serverName) throws IOException {
     return ConcurrentMapUtils.computeIfAbsentEx(adminSubs,
-      getStubKey(AdminService.getDescriptor().getName(), serverName, hostnameCanChange),
+      getStubKey(AdminService.getDescriptor().getName(), serverName),
       () -> createAdminServerStub(serverName));
   }
 

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

@@ -34,6 +34,8 @@
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.lang.reflect.UndeclaredThrowableException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
@@ -157,9 +159,6 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
   public static final String RETRIES_BY_SERVER_KEY = "hbase.client.retries.by.server";
   private static final Logger LOG = LoggerFactory.getLogger(ConnectionImplementation.class);
 
-  private static final String RESOLVE_HOSTNAME_ON_FAIL_KEY = "hbase.resolve.hostnames.on.failure";
-
-  private final boolean hostnamesCanChange;
   private final long pause;
   private final long pauseForCQTBE;// pause for CallQueueTooBigException, if specified
   // The mode tells if HedgedRead, LoadBalance mode is supported.
@@ -297,7 +296,6 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
 
     boolean shouldListen = conf.getBoolean(HConstants.STATUS_PUBLISHED,
         HConstants.STATUS_PUBLISHED_DEFAULT);
-    this.hostnamesCanChange = conf.getBoolean(RESOLVE_HOSTNAME_ON_FAIL_KEY, true);
     Class<? extends ClusterStatusListener.Listener> listenerClass =
         conf.getClass(ClusterStatusListener.STATUS_LISTENER_CLASS,
             ClusterStatusListener.DEFAULT_STATUS_LISTENER_CLASS,
@@ -476,7 +474,7 @@ public Hbck getHbck(ServerName masterServer) throws IOException {
       throw new RegionServerStoppedException(masterServer + " is dead.");
     }
     String key = getStubKey(MasterProtos.HbckService.BlockingInterface.class.getName(),
-      masterServer, this.hostnamesCanChange);
+      masterServer);
 
     return new HBaseHbck(
       (MasterProtos.HbckService.BlockingInterface) computeIfAbsentEx(stubs, key, () -> {
@@ -1242,7 +1240,7 @@ private MasterProtos.MasterService.BlockingInterface makeStubNoRetries()
       }
       // Use the security info interface name as our stub key
       String key =
-          getStubKey(MasterProtos.MasterService.getDescriptor().getName(), sn, hostnamesCanChange);
+          getStubKey(MasterProtos.MasterService.getDescriptor().getName(), sn);
       MasterProtos.MasterService.BlockingInterface stub =
           (MasterProtos.MasterService.BlockingInterface) computeIfAbsentEx(stubs, key, () -> {
             BlockingRpcChannel channel = rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
@@ -1290,8 +1288,7 @@ public AdminProtos.AdminService.BlockingInterface getAdmin(ServerName serverName
     if (isDeadServer(serverName)) {
       throw new RegionServerStoppedException(serverName + " is dead.");
     }
-    String key = getStubKey(AdminProtos.AdminService.BlockingInterface.class.getName(), serverName,
-      this.hostnamesCanChange);
+    String key = getStubKey(AdminProtos.AdminService.BlockingInterface.class.getName(), serverName);
     return (AdminProtos.AdminService.BlockingInterface) computeIfAbsentEx(stubs, key, () -> {
       BlockingRpcChannel channel =
           this.rpcClient.createBlockingRpcChannel(serverName, user, rpcTimeout);
@@ -1306,7 +1303,7 @@ public BlockingInterface getClient(ServerName serverName) throws IOException {
       throw new RegionServerStoppedException(serverName + " is dead.");
     }
     String key = getStubKey(ClientProtos.ClientService.BlockingInterface.class.getName(),
-      serverName, this.hostnamesCanChange);
+      serverName);
     return (ClientProtos.ClientService.BlockingInterface) computeIfAbsentEx(stubs, key, () -> {
       BlockingRpcChannel channel =
           this.rpcClient.createBlockingRpcChannel(serverName, user, rpcTimeout);

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

@@ -25,6 +25,7 @@
 import java.io.IOException;
 import java.lang.reflect.UndeclaredThrowableException;
 import java.net.InetAddress;
+import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.Arrays;
 import java.util.List;
@@ -216,32 +217,17 @@ public boolean isTableDisabled(TableName tableName) throws IOException {
   }
 
   /**
-   * Return retires + 1. The returned value will be in range [1, Integer.MAX_VALUE].
+   * Get a unique key for the rpc stub to the given server.
    */
-  static int retries2Attempts(int retries) {
-    return Math.max(1, retries == Integer.MAX_VALUE ? Integer.MAX_VALUE : retries + 1);
+  static String getStubKey(String serviceName, ServerName serverName) {
+    return String.format("%s@%s", serviceName, serverName);
   }
 
   /**
-   * Get a unique key for the rpc stub to the given server.
+   * Return retires + 1. The returned value will be in range [1, Integer.MAX_VALUE].
    */
-  static String getStubKey(String serviceName, ServerName serverName, boolean hostnameCanChange) {
-    // Sometimes, servers go down and they come back up with the same hostname but a different
-    // IP address. Force a resolution of the rsHostname by trying to instantiate an
-    // InetSocketAddress, and this way we will rightfully get a new stubKey.
-    // Also, include the hostname in the key so as to take care of those cases where the
-    // DNS name is different but IP address remains the same.
-    String hostname = serverName.getHostname();
-    int port = serverName.getPort();
-    if (hostnameCanChange) {
-      try {
-        InetAddress ip = InetAddress.getByName(hostname);
-        return serviceName + "@" + hostname + "-" + ip.getHostAddress() + ":" + port;
-      } catch (UnknownHostException e) {
-        LOG.warn("Can not resolve " + hostname + ", please check your network", e);
-      }
-    }
-    return serviceName + "@" + hostname + ":" + port;
+  static int retries2Attempts(int retries) {
+    return Math.max(1, retries == Integer.MAX_VALUE ? Integer.MAX_VALUE : retries + 1);
   }
 
   static void checkHasFamilies(Mutation mutation) {

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

@@ -66,6 +66,8 @@ public class MetricsConnection implements StatisticTrackable {
   private static final String HEAP_BASE = "heapOccupancy_";
   private static final String CACHE_BASE = "cacheDroppingExceptions_";
   private static final String UNKNOWN_EXCEPTION = "UnknownException";
+  private static final String NS_LOOKUPS = "nsLookups";
+  private static final String NS_LOOKUPS_FAILED = "nsLookupsFailed";
   private static final String CLIENT_SVC = ClientService.getDescriptor().getName();
 
   /** A container class for collecting details about the RPC call as it percolates. */
@@ -288,6 +290,8 @@ private static interface NewMetric<T> {
   protected final Counter hedgedReadWin;
   protected final Histogram concurrentCallsPerServerHist;
   protected final Histogram numActionsPerServerHist;
+  protected final Counter nsLookups;
+  protected final Counter nsLookupsFailed;
 
   // dynamic metrics
 
@@ -350,6 +354,8 @@ protected Ratio getRatio() {
       "concurrentCallsPerServer", scope));
     this.numActionsPerServerHist = registry.histogram(name(MetricsConnection.class,
       "numActionsPerServer", scope));
+    this.nsLookups = registry.counter(name(this.getClass(), NS_LOOKUPS, scope));
+    this.nsLookupsFailed = registry.counter(name(this.getClass(), NS_LOOKUPS_FAILED, scope));
 
     this.reporter = JmxReporter.forRegistry(this.registry).build();
     this.reporter.start();
@@ -518,4 +524,12 @@ public void incrCacheDroppingExceptions(Object exception) {
       (exception == null? UNKNOWN_EXCEPTION : exception.getClass().getSimpleName()),
       cacheDroppingExceptions, counterFactory).inc();
   }
+
+  public void incrNsLookups() {
+    this.nsLookups.inc();
+  }
+
+  public void incrNsLookupsFailed() {
+    this.nsLookupsFailed.inc();
+  }
 }