import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
/**
 * All initialization needed before we go register with Master.
 * @throws IOException
 * @throws InterruptedException
 */
private void preRegistrationInitialization() {
  try {
    initializeZooKeeper();

    clusterId = new ClusterId(zooKeeper, this);
    if (clusterId.hasId()) {
      conf.set(HConstants.CLUSTER_ID, clusterId.getId());
    }

    initializeThreads();
    int nbBlocks = conf.getInt("hbase.regionserver.nbreservationblocks", 4);
    for (int i = 0; i < nbBlocks; i++) {
      reservedSpace.add(new byte[HConstants.DEFAULT_SIZE_RESERVATION_BLOCK]);
    }

    this.rpcEngine = HBaseRPC.getProtocolEngine(conf);
  } catch (Throwable t) {
    // Call stop if error or process will stick around for ever since server
    // puts up non-daemon threads.
    this.rpcServer.stop();
    abort("Initialization of RS failed.  Hence aborting RS.", t);
  }
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
@BeforeClass
public static void setupBeforeClass() throws Exception {
  TEST_UTIL = new HBaseTestingUtility();
  Configuration conf = TEST_UTIL.getConfiguration();
  conf.set(HBaseRPC.RPC_ENGINE_PROP, SecureRpcEngine.class.getName());
  conf.set("hbase.coprocessor.region.classes",
      IdentityCoprocessor.class.getName());
  TEST_UTIL.startMiniCluster();
  HRegionServer rs = TEST_UTIL.getMiniHBaseCluster().getRegionServer(0);
  RpcServer server = rs.getRpcServer();
  assertTrue(server instanceof SecureServer);
  SecretManager mgr =
      ((SecureServer)server).getSecretManager();
  assertTrue(mgr instanceof AuthenticationTokenSecretManager);
  secretManager = (AuthenticationTokenSecretManager)mgr;
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
/**
 * All initialization needed before we go register with Master.
 *
 * @throws IOException
 * @throws InterruptedException
 */
private void preRegistrationInitialization(){
  try {
    initializeZooKeeper();

    clusterId = new ClusterId(zooKeeper, this);
    if(clusterId.hasId()) {
      conf.set(HConstants.CLUSTER_ID, clusterId.getId());
    }

    initializeThreads();
    int nbBlocks = conf.getInt("hbase.regionserver.nbreservationblocks", 4);
    for (int i = 0; i < nbBlocks; i++) {
      reservedSpace.add(new byte[HConstants.DEFAULT_SIZE_RESERVATION_BLOCK]);
    }

    this.rpcEngine = HBaseRPC.getProtocolEngine(conf);
  } catch (Throwable t) {
    // Call stop if error or process will stick around for ever since server
    // puts up non-daemon threads.
    this.rpcServer.stop();
    abort("Initialization of RS failed.  Hence aborting RS.", t);
  }
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
void close(boolean stopProxy) {
  if (this.closed) {
    return;
  }
  if (master != null) {
    if (stopProxy) {
      HBaseRPC.stopProxy(master);
    }
    master = null;
    masterChecked = false;
  }
  if (stopProxy) {
    for (HRegionInterface i : servers.values()) {
      HBaseRPC.stopProxy(i);
    }
  }
  this.servers.clear();
  if (this.zooKeeper != null) {
    LOG.info("Closed zookeeper sessionid=0x" +
      Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId()));
    this.zooKeeper.close();
    this.zooKeeper = null;
  }
  this.closed = true;
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
public void beforeCall() {
  this.startTime = EnvironmentEdgeManager.currentTimeMillis();
  int remaining = (int)(callTimeout - (this.startTime - this.globalStartTime));
  if (remaining < MIN_RPC_TIMEOUT) {
    // If there is no time left, we're trying anyway. It's too late.
    // 0 means no timeout, and it's not the intent here. So we secure both cases by
    // resetting to the minimum.
    remaining = MIN_RPC_TIMEOUT;
  }
  HBaseRPC.setRpcTimeout(remaining);
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
@Test
public void testRPCException() throws Exception {
  HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
  TEST_UTIL.startMiniZKCluster();
  Configuration conf = TEST_UTIL.getConfiguration();
  conf.set(HConstants.MASTER_PORT, "0");

  HMaster hm = new HMaster(conf);

  ServerName sm = hm.getServerName();
  InetSocketAddress isa = new InetSocketAddress(sm.getHostname(), sm.getPort());
  RpcEngine rpcEngine = null;
  try {
    rpcEngine = HBaseRPC.getProtocolEngine(conf);
    HMasterInterface inf = rpcEngine.getProxy(
        HMasterInterface.class,  HMasterInterface.VERSION, isa, conf, 100 * 10);
    inf.isMasterRunning();
    fail();
  } catch (RemoteException ex) {
    assertTrue(ex.getMessage().startsWith(
        "org.apache.hadoop.hbase.ipc.ServerNotRunningYetException: Server is not running yet"));
  } catch (Throwable t) {
    fail("Unexpected throwable: " + t);
  } finally {
    if (rpcEngine != null) {
      rpcEngine.close();
    }
  }
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
@Override
public ProtocolSignature getProtocolSignature(String protocol,
    long clientVersion, int clientMethodsHash) throws IOException {
  if (AccessControllerProtocol.class.getName().equals(protocol)) {
    return new ProtocolSignature(PROTOCOL_VERSION, null);
  }
  throw new HBaseRPC.UnknownProtocolException(
      "Unexpected protocol requested: "+protocol);
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
/**
 * Either the passed <code>isa</code> is null or <code>hostname</code>
 * can be but not both.
 * @param hostname
 * @param port
 * @param isa
 * @param master
 * @return Proxy.
 * @throws IOException
 */
HRegionInterface getHRegionConnection(final String hostname, final int port,
    final InetSocketAddress isa, final boolean master)
throws IOException {
  if (master) getMaster();
  HRegionInterface server;
  String rsName = null;
  if (isa != null) {
    rsName = Addressing.createHostAndPortStr(isa.getHostName(),
        isa.getPort());
  } else {
    rsName = Addressing.createHostAndPortStr(hostname, port);
  }
  ensureZookeeperTrackers();
  // See if we already have a connection (common case)
  server = this.servers.get(rsName);
  if (server == null) {
    // create a unique lock for this RS (if necessary)
    this.connectionLock.putIfAbsent(rsName, rsName);
    // get the RS lock
    synchronized (this.connectionLock.get(rsName)) {
      // do one more lookup in case we were stalled above
      server = this.servers.get(rsName);
      if (server == null) {
        try {
          // Only create isa when we need to.
          InetSocketAddress address = isa != null? isa:
            new InetSocketAddress(hostname, port);
          // definitely a cache miss. establish an RPC for this RS
          server = HBaseRPC.waitForProxy(this.rpcEngine,
              serverInterfaceClass, HRegionInterface.VERSION,
              address, this.conf,
              this.maxRPCAttempts, this.rpcTimeout, this.rpcTimeout);
          this.servers.put(Addressing.createHostAndPortStr(
              address.getHostName(), address.getPort()), server);
        } catch (RemoteException e) {
          LOG.warn("RemoteException connecting to RS", e);
          // Throw what the RemoteException was carrying.
          throw e.unwrapRemoteException();
        }
      }
    }
  }
  return server;
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
public void afterCall() {
  HBaseRPC.resetRpcTimeout();
  this.endTime = EnvironmentEdgeManager.currentTimeMillis();
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
/**
 * Get the current master from ZooKeeper and open the RPC connection to it. Method will block
 * until a master is available. You can break from this block by requesting the server stop.
 * @return master + port, or null if server has been stopped
 */
private ServerName getMaster() {
  ServerName masterServerName = null;
  long previousLogTime = 0;
  HMasterRegionInterface master = null;
  InetSocketAddress masterIsa = null;
  while (keepLooping() && master == null) {
    masterServerName = this.masterAddressManager.getMasterAddress();
    if (masterServerName == null) {
      if (!keepLooping()) {
        // give up with no connection.
        LOG.debug("No master found and cluster is stopped; bailing out");
        return null;
      }
      LOG.debug("No master found; retry");
      previousLogTime = System.currentTimeMillis();

      sleeper.sleep();
      continue;
    }

    masterIsa = new InetSocketAddress(masterServerName.getHostname(), masterServerName.getPort());

    LOG.info("Attempting connect to Master server at " + masterServerName);
    try {
      // Do initial RPC setup. The final argument indicates that the RPC
      // should retry indefinitely.
      master =
          HBaseRPC.waitForProxy(this.rpcEngine, HMasterRegionInterface.class,
            HMasterRegionInterface.VERSION, masterIsa, this.conf, -1, this.rpcTimeout,
            this.rpcTimeout);
    } catch (IOException e) {
      e = e instanceof RemoteException ? ((RemoteException) e).unwrapRemoteException() : e;
      if (e instanceof ServerNotRunningYetException) {
        if (System.currentTimeMillis() > (previousLogTime + 1000)) {
          LOG.info("Master isn't available yet, retrying");
          previousLogTime = System.currentTimeMillis();
        }
      } else {
        if (System.currentTimeMillis() > (previousLogTime + 1000)) {
          LOG.warn("Unable to connect to master. Retrying. Error was:", e);
          previousLogTime = System.currentTimeMillis();
        }
      }
      try {
        Thread.sleep(200);
      } catch (InterruptedException ignored) {
      }
    }
  }
  LOG.info("Connected to master at " + masterIsa);
  this.hbaseMaster = master;
  return masterServerName;
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
/**
 * The main run method of TestHLogBench
 */
public int run(String argv[]) throws Exception {

  int exitCode = -1;
  int i = 0;

  // verify that we have enough command line parameters
  if (argv.length < 4) {
    printUsage("");
    return exitCode;
  }

  // initialize LogBench
  try {
    init();
  } catch (HBaseRPC.VersionMismatch v) {
    LOG.warn("Version Mismatch between client and server" +
             "... command aborted.");
    return exitCode;
  } catch (IOException e) {
    LOG.warn("Bad connection to FS. command aborted.");
    return exitCode;
  }

  try {
    for (; i < argv.length; i++) {
      if ("-numThreads".equals(argv[i])) {
        i++;
        this.numThreads = Integer.parseInt(argv[i]);
      } else if ("-numIterationsPerThread".equals(argv[i])) {
        i++;
        this.numIterationsPerThread = Integer.parseInt(argv[i]);
      } else if ("-path".equals(argv[i])) {
        // get an absolute path using the default file system
        i++;
        this.regionRootDir = new Path(argv[i]);
        this.regionRootDir = regionRootDir.makeQualified(this.fs);
      } else if ("-nosync".equals(argv[i])) {
        this.appendNoSync = true;
      } else {
        printUsage(argv[i]);
        return exitCode;
      }
    }
  } catch (NumberFormatException nfe) {
    LOG.warn("Illegal numThreads or numIterationsPerThread, " +
             " a positive integer expected");
    throw nfe;
  }
  go();
  return 0;
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
/**
 * Get the current master from ZooKeeper and open the RPC connection to it.
 *
 * Method will block until a master is available. You can break from this
 * block by requesting the server stop.
 *
 * @return master + port, or null if server has been stopped
 */
private ServerName getMaster() {
  ServerName masterServerName = null;
  long previousLogTime = 0;
  HMasterRegionInterface master = null;
  InetSocketAddress masterIsa = null;
  while (keepLooping() && master == null) {
    masterServerName = this.masterAddressManager.getMasterAddress();
    if (masterServerName == null) {
      if (!keepLooping()) {
        // give up with no connection.
        LOG.debug("No master found and cluster is stopped; bailing out");
        return null;
      }
      LOG.debug("No master found; retry");
      previousLogTime = System.currentTimeMillis();

      sleeper.sleep();
      continue;
    }

    masterIsa =
      new InetSocketAddress(masterServerName.getHostname(), masterServerName.getPort());

    LOG.info("Attempting connect to Master server at " + masterServerName);
    try {
      // Do initial RPC setup. The final argument indicates that the RPC
      // should retry indefinitely.
      master = HBaseRPC.waitForProxy(this.rpcEngine,
          HMasterRegionInterface.class, HMasterRegionInterface.VERSION,
          masterIsa, this.conf, -1,
          this.rpcTimeout, this.rpcTimeout);
    } catch (IOException e) {
      e = e instanceof RemoteException ?
          ((RemoteException)e).unwrapRemoteException() : e;
      if (e instanceof ServerNotRunningYetException) {
        if (System.currentTimeMillis() > (previousLogTime+1000)){
          LOG.info("Master isn't available yet, retrying");
          previousLogTime = System.currentTimeMillis();
        }
      } else {
        if (System.currentTimeMillis() > (previousLogTime + 1000)) {
          LOG.warn("Unable to connect to master. Retrying. Error was:", e);
          previousLogTime = System.currentTimeMillis();
        }
      }
      try {
        Thread.sleep(200);
      } catch (InterruptedException ignored) {
      }
    }
  }
  LOG.info("Connected to master at " + masterIsa);
  this.hbaseMaster = master;
  return masterServerName;
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
public void beforeCall() {
  HBaseRPC.setRpcTimeout(this.callTimeout);
  this.startTime = System.currentTimeMillis();
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
public void afterCall() {
  HBaseRPC.resetRpcTimeout();
  this.endTime = System.currentTimeMillis();
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
/**
 * Initializes the HMaster. The steps are as follows:
 * <p>
 * <ol>
 * <li>Initialize HMaster RPC and address
 * <li>Connect to ZooKeeper.
 * </ol>
 * <p>
 * Remaining steps of initialization occur in {@link #run()} so that they
 * run in their own thread rather than within the context of the constructor.
 * @throws InterruptedException
 */
public HMaster(final Configuration conf)
throws IOException, KeeperException, InterruptedException {
  this.conf = new Configuration(conf);
  // Disable the block cache on the master
  this.conf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0.0f);
  // Set how many times to retry talking to another server over HConnection.
  HConnectionManager.setServerSideHConnectionRetries(this.conf, LOG);
  // Server to handle client requests.
  String hostname = DNS.getDefaultHost(
    conf.get("hbase.master.dns.interface", "default"),
    conf.get("hbase.master.dns.nameserver", "default"));
  int port = conf.getInt(HConstants.MASTER_PORT, HConstants.DEFAULT_MASTER_PORT);
  // Creation of a HSA will force a resolve.
  InetSocketAddress initialIsa = new InetSocketAddress(hostname, port);
  if (initialIsa.getAddress() == null) {
    throw new IllegalArgumentException("Failed resolve of " + this.isa);
  }
  int numHandlers = conf.getInt("hbase.master.handler.count",
    conf.getInt("hbase.regionserver.handler.count", 25));
  this.rpcServer = HBaseRPC.getServer(this,
    new Class<?>[]{HMasterInterface.class, HMasterRegionInterface.class},
      initialIsa.getHostName(), // BindAddress is IP we got for this server.
      initialIsa.getPort(),
      numHandlers,
      0, // we dont use high priority handlers in master
      conf.getBoolean("hbase.rpc.verbose", false), conf,
      0); // this is a DNC w/o high priority handlers
  // Set our address.
  this.isa = this.rpcServer.getListenerAddress();
  this.serverName = new ServerName(this.isa.getHostName(),
    this.isa.getPort(), System.currentTimeMillis());
  this.rsFatals = new MemoryBoundedLogMessageBuffer(
      conf.getLong("hbase.master.buffer.for.rs.fatals", 1*1024*1024));

  // initialize server principal (if using secure Hadoop)
  User.login(conf, "hbase.master.keytab.file",
    "hbase.master.kerberos.principal", this.isa.getHostName());

  // set the thread name now we have an address
  setName(MASTER + "-" + this.serverName.toString());

  Replication.decorateMasterConfiguration(this.conf);

  // Hack! Maps DFSClient => Master for logs.  HDFS made this
  // config param for task trackers, but we can piggyback off of it.
  if (this.conf.get("mapred.task.id") == null) {
    this.conf.set("mapred.task.id", "hb_m_" + this.serverName.toString());
  }

  this.zooKeeper = new ZooKeeperWatcher(conf, MASTER + ":" + isa.getPort(), this, true);
  this.rpcServer.startThreads();
  this.metrics = new MasterMetrics(getServerName().toString());
}
 

import org.apache.hadoop.hbase.ipc.HBaseRPC; //导入依赖的package包/类
/**
 * Starts a HRegionServer at the default location
 *
 * @param conf
 * @throws IOException
 * @throws InterruptedException
 */
public HRegionServer(Configuration conf)
throws IOException, InterruptedException {
  this.fsOk = true;
  this.conf = conf;
  // Set how many times to retry talking to another server over HConnection.
  HConnectionManager.setServerSideHConnectionRetries(this.conf, LOG);
  this.isOnline = false;
  checkCodecs(this.conf);

  // Config'ed params
  this.numRetries = conf.getInt("hbase.client.retries.number", 10);
  this.threadWakeFrequency = conf.getInt(HConstants.THREAD_WAKE_FREQUENCY,
    10 * 1000);
  this.msgInterval = conf.getInt("hbase.regionserver.msginterval", 3 * 1000);

  this.sleeper = new Sleeper(this.msgInterval, this);

  this.maxScannerResultSize = conf.getLong(
    HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
    HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);

  this.numRegionsToReport = conf.getInt(
    "hbase.regionserver.numregionstoreport", 10);

  this.rpcTimeout = conf.getInt(
    HConstants.HBASE_RPC_TIMEOUT_KEY,
    HConstants.DEFAULT_HBASE_RPC_TIMEOUT);

  this.abortRequested = false;
  this.stopped = false;

  // Server to handle client requests.
  String hostname = DNS.getDefaultHost(
    conf.get("hbase.regionserver.dns.interface", "default"),
    conf.get("hbase.regionserver.dns.nameserver", "default"));
  int port = conf.getInt(HConstants.REGIONSERVER_PORT,
    HConstants.DEFAULT_REGIONSERVER_PORT);
  // Creation of a HSA will force a resolve.
  InetSocketAddress initialIsa = new InetSocketAddress(hostname, port);
  if (initialIsa.getAddress() == null) {
    throw new IllegalArgumentException("Failed resolve of " + initialIsa);
  }
  this.rpcServer = HBaseRPC.getServer(this,
    new Class<?>[]{HRegionInterface.class, HBaseRPCErrorHandler.class,
      OnlineRegions.class},
      initialIsa.getHostName(), // BindAddress is IP we got for this server.
      initialIsa.getPort(),
      conf.getInt("hbase.regionserver.handler.count", 10),
      conf.getInt("hbase.regionserver.metahandler.count", 10),
      conf.getBoolean("hbase.rpc.verbose", false),
      conf, QOS_THRESHOLD);
  // Set our address.
  this.isa = this.rpcServer.getListenerAddress();

  this.rpcServer.setErrorHandler(this);
  this.rpcServer.setQosFunction(new QosFunction());
  this.startcode = System.currentTimeMillis();

  // login the server principal (if using secure Hadoop)
  User.login(this.conf, "hbase.regionserver.keytab.file",
    "hbase.regionserver.kerberos.principal", this.isa.getHostName());
  regionServerAccounting = new RegionServerAccounting();
  cacheConfig = new CacheConfig(conf);
}