HBASE-18095: Zookeeper-less client connection implementation

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

@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+import java.io.IOException;
+
+/**
+ * Thrown by master when meta region locations are not cached whatever reason.
+ * Client is expected to retry when running into this.
+ */
+@InterfaceAudience.Private
+public class MetaRegionsNotAvailableException extends IOException  {
+    private static final long serialVersionUID = (1L << 14) - 1L;
+
+    public MetaRegionsNotAvailableException(String msg) {
+        super(msg);
+    }
+}

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

@@ -394,4 +394,7 @@ public Hbck getHbck(ServerName masterServer) throws IOException {
   Optional<MetricsConnection> getConnectionMetrics() {
     return metrics;
   }
+
+  @VisibleForTesting
+  AsyncRegistry getRegistry() { return registry; }
 }

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

@@ -26,7 +26,6 @@
 
 /**
  * Implementations hold cluster information such as this cluster's id, location of hbase:meta, etc..
- * All stuffs that may be related to zookeeper at client side are placed here.
  * <p>
  * Internal use only.
  */
@@ -45,21 +44,11 @@ interface AsyncRegistry extends Closeable {
    */
   CompletableFuture<String> getClusterId();
 
-  /**
-   * Get the number of 'running' regionservers.
-   */
-  CompletableFuture<Integer> getCurrentNrHRS();
-
   /**
    * Get the address of HMaster.
    */
   CompletableFuture<ServerName> getMasterAddress();
 
-  /**
-   * Get the info port of HMaster.
-   */
-  CompletableFuture<Integer> getMasterInfoPort();
-
   /**
    * Closes this instance and releases any system resources associated with it
    */

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

@@ -27,7 +27,7 @@
 @InterfaceAudience.Private
 final class AsyncRegistryFactory {
 
-  static final String REGISTRY_IMPL_CONF_KEY = "hbase.client.registry.impl";
+  public static final String REGISTRY_IMPL_CONF_KEY = "hbase.client.registry.impl";
 
   private AsyncRegistryFactory() {
   }

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

@@ -0,0 +1,236 @@
+/**
+ * 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.client;
+
+import static org.apache.hadoop.hbase.HConstants.DEFAULT_HBASE_RPC_TIMEOUT;
+import static org.apache.hadoop.hbase.HConstants.HBASE_RPC_TIMEOUT_KEY;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.MasterNotRunningException;
+import org.apache.hadoop.hbase.MetaRegionsNotAvailableException;
+import org.apache.hadoop.hbase.RegionLocations;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.ipc.HBaseRpcController;
+import org.apache.hadoop.hbase.ipc.RpcClient;
+import org.apache.hadoop.hbase.ipc.RpcClientFactory;
+import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetClusterIdRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetClusterIdResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetMetaLocationsRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetMetaLocationsResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsActiveRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsActiveResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MasterService;
+import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
+import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
+
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Fetches the meta information directly from HMaster by making relevant RPCs. HMaster RPC end
+ * points are looked up via client configuration 'hbase.client.asyncregistry.masteraddrs' as
+ * comma separated list of host:port values. Should be set in conjunction with
+ * 'hbase.client.registry.impl' set to this class impl.
+ *
+ * The class does not cache anything. It is the responsibility of the callers to cache and
+ * avoid repeated requests.
+ */
+@InterfaceAudience.Private
+public class HMasterAsyncRegistry implements AsyncRegistry {
+  private static final Logger LOG = LoggerFactory.getLogger(HMasterAsyncRegistry.class);
+  public static final String CONF_KEY = "hbase.client.asyncregistry.masteraddrs";
+  private static final String DEFAULT_HOST_PORT = "localhost:" + HConstants.DEFAULT_MASTER_PORT;
+
+  // Parsed list of host and ports for masters from hbase-site.xml
+  private final List<ServerName> masterServers;
+  final Configuration conf;
+  // RPC client used to talk to the master servers. This uses a stand alone RpcClient instance
+  // because AsyncRegistry is created prior to creating a cluster Connection. The client is torn
+  // down in close().
+  final RpcClient rpcClient;
+  final int rpcTimeout;
+
+  public HMasterAsyncRegistry(Configuration config) {
+    masterServers = new ArrayList<>();
+    conf = config;
+    parseHortPorts();
+    // Passing the default cluster ID means that the token based authentication does not work for
+    // certain client implementations.
+    // TODO(bharathv): Figure out a way to fetch the CLUSTER ID using a non authenticated way.
+    rpcClient = RpcClientFactory.createClient(conf, HConstants.CLUSTER_ID_DEFAULT);
+    rpcTimeout = (int) Math.min(Integer.MAX_VALUE,
+        TimeUnit.MILLISECONDS.toNanos(conf.getLong(HBASE_RPC_TIMEOUT_KEY,
+            DEFAULT_HBASE_RPC_TIMEOUT)));
+  }
+
+  private void parseHortPorts() {
+    String hostPorts = conf.get(CONF_KEY, DEFAULT_HOST_PORT);
+    for (String hostPort: hostPorts.split(",")) {
+      masterServers.add(ServerName.valueOf(hostPort, ServerName.NON_STARTCODE));
+    }
+    Preconditions.checkArgument(!masterServers.isEmpty(), String.format("%s is empty", CONF_KEY));
+    // Randomize so that not every client sends requests in the same order.
+    Collections.shuffle(masterServers);
+  }
+
+  /**
+   * Util that generates a master stub for a given ServerName.
+   */
+  private MasterService.BlockingInterface getMasterStub(ServerName server) throws IOException  {
+    return MasterService.newBlockingStub(
+        rpcClient.createBlockingRpcChannel(server, User.getCurrent(), rpcTimeout));
+  }
+
+  /**
+   * Blocking RPC to fetch the meta region locations using one of the masters from the parsed list.
+   */
+  private RegionLocations getMetaRegionLocationsHelper() throws MetaRegionsNotAvailableException {
+    List<HBaseProtos.RegionLocation> result = null;
+    for (ServerName sname: masterServers) {
+      try {
+        MasterService.BlockingInterface stub = getMasterStub(sname);
+        HBaseRpcController rpcController = RpcControllerFactory.instantiate(conf).newController();
+        GetMetaLocationsResponse resp = stub.getMetaLocations(rpcController,
+            GetMetaLocationsRequest.getDefaultInstance());
+        result = resp.getLocationsList();
+      } catch (Exception e) {
+        LOG.warn("Error fetch meta locations from master {}. Trying others.", sname, e);
+      }
+    }
+    if (result == null || result.isEmpty()) {
+      throw new MetaRegionsNotAvailableException(String.format(
+          "Meta locations not found. Probed masters: %s", conf.get(CONF_KEY, DEFAULT_HOST_PORT)));
+    }
+    List<HRegionLocation> deserializedResult = new ArrayList<>();
+    result.stream().forEach(
+        location -> deserializedResult.add(ProtobufUtil.toRegionLocation(location)));
+    return new RegionLocations(deserializedResult);
+  }
+
+  /**
+   * Picks the first master entry from 'masterHortPorts' to fetch the meta region locations.
+   */
+  @Override
+  public CompletableFuture<RegionLocations> getMetaRegionLocation() {
+    CompletableFuture<RegionLocations> result = new CompletableFuture<>();
+    CompletableFuture.runAsync(() -> {
+      try {
+        result.complete(this.getMetaRegionLocationsHelper());
+      } catch (Exception e) {
+        result.completeExceptionally(e);
+      }
+    });
+    return result;
+  }
+
+  /**
+   * Blocking RPC to get the cluster ID from the parsed master list. Returns null if no active
+   * master found.
+   */
+  private String getClusterIdHelper() throws  MasterNotRunningException {
+    // Loop through all the masters serially. We could be hitting some standby masters which cannot
+    // process this RPC, so we just skip them.
+    for (ServerName sname: masterServers) {
+      try {
+        MasterService.BlockingInterface stub = getMasterStub(sname);
+        HBaseRpcController rpcController = RpcControllerFactory.instantiate(conf).newController();
+        GetClusterIdResponse resp =
+            stub.getClusterId(rpcController, GetClusterIdRequest.getDefaultInstance());
+        return resp.getClusterId();
+      } catch (IOException e) {
+        LOG.warn("Error fetching cluster ID from master: {}", sname, e);
+      } catch (ServiceException e) {
+        // This is probably a standby master, can be ignored.
+        LOG.debug("Error fetching cluster ID from server: {}" , sname, e);
+      }
+    }
+    // If it comes to this point, no active master could be found.
+    throw new MasterNotRunningException(String.format(
+        "No active master found. Probed masters: %s", conf.get(CONF_KEY, DEFAULT_HOST_PORT)));
+  }
+
+  @Override
+  public CompletableFuture<String> getClusterId() {
+    CompletableFuture<String> result = new CompletableFuture<>();
+    CompletableFuture.runAsync(() -> {
+      try {
+        result.complete(this.getClusterIdHelper());
+      } catch (Exception e) {
+        result.completeExceptionally(e);
+      }
+    });
+    return result;
+  }
+
+  /**
+   * Blocking RPC to get the active master address from the parsed list of master servers.
+   */
+  private ServerName getMasterAddressHelper() throws MasterNotRunningException  {
+    for (ServerName sname: masterServers) {
+      try {
+        MasterService.BlockingInterface stub = getMasterStub(sname);
+        HBaseRpcController rpcController = RpcControllerFactory.instantiate(conf).newController();
+        IsActiveResponse resp = stub.isActive(rpcController, IsActiveRequest.getDefaultInstance());
+        if (resp.getIsMasterActive()) {
+          return ServerName.valueOf(sname.getHostname(), sname.getPort(), resp.getStartCode());
+        }
+      } catch (Exception e) {
+
+      }
+    }
+    throw new MasterNotRunningException(String.format("No active master found. Probed masters: %s",
+        conf.get(CONF_KEY, DEFAULT_HOST_PORT)));
+  }
+
+  /**
+   * @return the active master among the configured master addresses in 'masterHortPorts'.
+   */
+  @Override
+  public CompletableFuture<ServerName> getMasterAddress() {
+    CompletableFuture<ServerName> result = new CompletableFuture<>();
+    CompletableFuture.runAsync(() -> {
+      try {
+        result.complete(this.getMasterAddressHelper());
+      } catch (Exception e) {
+        result.completeExceptionally(e);
+      }
+    });
+    return result;
+  }
+
+  @Override
+  public void close() {
+    if (rpcClient != null) {
+      rpcClient.close();
+    }
+  }
+}

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

@@ -210,11 +210,6 @@ public CompletableFuture<RegionLocations> getMetaRegionLocation() {
     return future;
   }
 
-  @Override
-  public CompletableFuture<Integer> getCurrentNrHRS() {
-    return zk.exists(znodePaths.rsZNode).thenApply(s -> s != null ? s.getNumChildren() : 0);
-  }
-
   private static ZooKeeperProtos.Master getMasterProto(byte[] data) throws IOException {
     if (data == null || data.length == 0) {
       return null;
@@ -237,12 +232,6 @@ public CompletableFuture<ServerName> getMasterAddress() {
         });
   }
 
-  @Override
-  public CompletableFuture<Integer> getMasterInfoPort() {
-    return getAndConvert(znodePaths.masterAddressZNode, ZKAsyncRegistry::getMasterProto)
-        .thenApply(proto -> proto != null ? proto.getInfoPort() : 0);
-  }
-
   @Override
   public void close() {
     zk.close();

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

@@ -3188,6 +3188,7 @@ public static org.apache.hadoop.hbase.client.RegionInfo toRegionInfo(final HBase
   }
 
   public static HBaseProtos.RegionLocation toRegionLocation(HRegionLocation loc) {
+    if (loc == null) return null;
     HBaseProtos.RegionLocation.Builder builder = HBaseProtos.RegionLocation.newBuilder();
     builder.setRegionInfo(toRegionInfo(loc.getRegion()));
     if (loc.getServerName() != null) {

hbase-client/src/main/java/org/apache/hadoop/hbase/zookeeper/ZNodePaths.java

@@ -182,6 +182,18 @@ public String getZNodeForReplica(int replicaId) {
         .orElseGet(() -> metaReplicaZNodes.get(DEFAULT_REPLICA_ID) + "-" + replicaId);
   }
 
+  /**
+   * Parses the meta replicaId from the passed path.
+   * @param path the name of the full path which includes baseZNode.
+   * @return replicaId
+   */
+  public int getMetaReplicaIdFromPath(String path) {
+    // Extract the znode from path. The prefix is of the following format.
+    // baseZNode + PATH_SEPARATOR.
+    int prefixLen = baseZNode.length() + 1;
+    return getMetaReplicaIdFromZnode(path.substring(prefixLen));
+  }
+
   /**
    * Parse the meta replicaId from the passed znode
    * @param znode the name of the znode, does not include baseZNode

hbase-client/src/test/java/org/apache/hadoop/hbase/client/DoNothingAsyncRegistry.java

@@ -43,21 +43,11 @@ public CompletableFuture<String> getClusterId() {
     return CompletableFuture.completedFuture(null);
   }
 
-  @Override
-  public CompletableFuture<Integer> getCurrentNrHRS() {
-    return CompletableFuture.completedFuture(0);
-  }
-
   @Override
   public CompletableFuture<ServerName> getMasterAddress() {
     return CompletableFuture.completedFuture(null);
   }
 
-  @Override
-  public CompletableFuture<Integer> getMasterInfoPort() {
-    return CompletableFuture.completedFuture(0);
-  }
-
   @Override
   public void close() {
   }