@Override
  @QosPriority(priority = HConstants.ADMIN_QOS)
  public OpenRegionResponse openRegion(final RpcController controller,
    final OpenRegionRequest request) throws ServiceException {
    requestCount.increment();
    throwOnWrongStartCode(request);

    OpenRegionResponse.Builder builder = OpenRegionResponse.newBuilder();
    final int regionCount = request.getOpenInfoCount();
    final Map<TableName, TableDescriptor> htds = new HashMap<>(regionCount);
    final boolean isBulkAssign = regionCount > 1;
    try {
      checkOpen();
    } catch (IOException ie) {
      TableName tableName = null;
      if (regionCount == 1) {
        org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.RegionInfo ri =
          request.getOpenInfo(0).getRegion();
        if (ri != null) {
          tableName = ProtobufUtil.toTableName(ri.getTableName());
        }
      }
      if (!TableName.META_TABLE_NAME.equals(tableName)) {
        throw new ServiceException(ie);
      }
      // We are assigning meta, wait a little for regionserver to finish initialization.
      // Default to quarter of RPC timeout
      int timeout = server.getConfiguration().getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
        HConstants.DEFAULT_HBASE_RPC_TIMEOUT) >> 2;
      long endTime = EnvironmentEdgeManager.currentTime() + timeout;
      synchronized (server.online) {
        try {
          while (
            EnvironmentEdgeManager.currentTime() <= endTime && !server.isStopped()
              && !server.isOnline()
          ) {
            server.online.wait(server.getMsgInterval());
          }
          checkOpen();
        } catch (InterruptedException t) {
          Thread.currentThread().interrupt();
          throw new ServiceException(t);
        } catch (IOException e) {
          throw new ServiceException(e);
        }
      }
    }

    long masterSystemTime = request.hasMasterSystemTime() ? request.getMasterSystemTime() : -1;

    for (RegionOpenInfo regionOpenInfo : request.getOpenInfoList()) {
      final RegionInfo region = ProtobufUtil.toRegionInfo(regionOpenInfo.getRegion());
      TableDescriptor htd;
      try {
        String encodedName = region.getEncodedName();
        byte[] encodedNameBytes = region.getEncodedNameAsBytes();
        final HRegion onlineRegion = server.getRegion(encodedName);
        if (onlineRegion != null) {
          // The region is already online. This should not happen any more.
          String error = "Received OPEN for the region:" + region.getRegionNameAsString()
            + ", which is already online";
          LOG.warn(error);
          // server.abort(error);
          // throw new IOException(error);
          builder.addOpeningState(RegionOpeningState.OPENED);
          continue;
        }
        LOG.info("Open " + region.getRegionNameAsString());

        final Boolean previous =
          server.getRegionsInTransitionInRS().putIfAbsent(encodedNameBytes, Boolean.TRUE);

        if (Boolean.FALSE.equals(previous)) {
          if (server.getRegion(encodedName) != null) {
            // There is a close in progress. This should not happen any more.
            String error = "Received OPEN for the region:" + region.getRegionNameAsString()
              + ", which we are already trying to CLOSE";
            server.abort(error);
            throw new IOException(error);
          }
          server.getRegionsInTransitionInRS().put(encodedNameBytes, Boolean.TRUE);
        }

        if (Boolean.TRUE.equals(previous)) {
          // An open is in progress. This is supported, but let's log this.
          LOG.info("Receiving OPEN for the region:" + region.getRegionNameAsString()
            + ", which we are already trying to OPEN"
            + " - ignoring this new request for this region.");
        }

        // We are opening this region. If it moves back and forth for whatever reason, we don't
        // want to keep returning the stale moved record while we are opening/if we close again.
        server.removeFromMovedRegions(region.getEncodedName());

        if (previous == null || !previous.booleanValue()) {
          htd = htds.get(region.getTable());
          if (htd == null) {
            htd = server.getTableDescriptors().get(region.getTable());
            htds.put(region.getTable(), htd);
          }
          if (htd == null) {
            throw new IOException("Missing table descriptor for " + region.getEncodedName());
          }
          // If there is no action in progress, we can submit a specific handler.
          // Need to pass the expected version in the constructor.
          if (server.getExecutorService() == null) {
            LOG.info("No executor executorService; skipping open request");
          } else {
            if (region.isMetaRegion()) {
              server.getExecutorService()
                .submit(new OpenMetaHandler(server, server, region, htd, masterSystemTime));
            } else {
              if (regionOpenInfo.getFavoredNodesCount() > 0) {
                server.updateRegionFavoredNodesMapping(region.getEncodedName(),
                  regionOpenInfo.getFavoredNodesList());
              }
              if (htd.getPriority() >= HConstants.ADMIN_QOS || region.getTable().isSystemTable()) {
                server.getExecutorService().submit(
                  new OpenPriorityRegionHandler(server, server, region, htd, masterSystemTime));
              } else {
                server.getExecutorService()
                  .submit(new OpenRegionHandler(server, server, region, htd, masterSystemTime));
              }
            }
          }
        }

        builder.addOpeningState(RegionOpeningState.OPENED);
      } catch (IOException ie) {
        LOG.warn("Failed opening region " + region.getRegionNameAsString(), ie);
        if (isBulkAssign) {
          builder.addOpeningState(RegionOpeningState.FAILED_OPENING);
        } else {
          throw new ServiceException(ie);
        }
      }
    }
    return builder.build();
  }
Open asynchronously a region or a set of regions on the region server. The opening is
coordinated by ZooKeeper, and this method requires the znode to be created before being called.
As a consequence, this method should be called only from the master.
<p>
Different manages states for the region are:
</p>
<ul>
<li>region not opened: the region opening will start asynchronously.</li>
<li>a close is already in progress: this is considered as an error.</li>
<li>an open is already in progress: this new open request will be ignored. This is important
because the Master can do multiple requests if it crashes.</li>
<li>the region is already opened: this new open request will be ignored.</li>
</ul>
<p>
Bulk assign: If there are more than 1 region to open, it will be considered as a bulk assign.
For a single region opening, errors are sent through a ServiceException. For bulk assign,
errors are put in the response as FAILED_OPENING.
</p>
@param controller the RPC controller
@param request    the request
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java