private static ThriftServer startBinaryMetastore(int port, HadoopThriftAuthBridge bridge,
      Configuration conf) throws Throwable {
    // Server will create new threads up to max as necessary. After an idle
    // period, it will destroy threads to keep the number of threads in the
    // pool to min.
    long maxMessageSize = MetastoreConf.getLongVar(conf, ConfVars.SERVER_MAX_MESSAGE_SIZE);
    int minWorkerThreads = MetastoreConf.getIntVar(conf, ConfVars.SERVER_MIN_THREADS);
    int maxWorkerThreads = MetastoreConf.getIntVar(conf, ConfVars.SERVER_MAX_THREADS);
    boolean tcpKeepAlive = MetastoreConf.getBoolVar(conf, ConfVars.TCP_KEEP_ALIVE);
    boolean useCompactProtocol = MetastoreConf.getBoolVar(conf, ConfVars.USE_THRIFT_COMPACT_PROTOCOL);
    boolean useSSL = MetastoreConf.getBoolVar(conf, ConfVars.USE_SSL);
    HMSHandler baseHandler = new HMSHandler("new db based metaserver", conf);
    AuthFactory authFactory = new AuthFactory(bridge, conf, baseHandler);
    useSasl = authFactory.isSASLWithKerberizedHadoop();

    if (useSasl) {
      // we are in secure mode. Login using keytab
      String kerberosName = SecurityUtil
          .getServerPrincipal(MetastoreConf.getVar(conf, ConfVars.KERBEROS_PRINCIPAL), "0.0.0.0");
      String keyTabFile = MetastoreConf.getVar(conf, ConfVars.KERBEROS_KEYTAB_FILE);
      UserGroupInformation.loginUserFromKeytab(kerberosName, keyTabFile);
      saslServer = authFactory.getSaslServer();
      delegationTokenManager = authFactory.getDelegationTokenManager();
    }

    TProcessor processor;
    TTransportFactory transFactory = authFactory.getAuthTransFactory(useSSL, conf);
    final TProtocolFactory protocolFactory;
    final TProtocolFactory inputProtoFactory;
    if (useCompactProtocol) {
      protocolFactory = new TCompactProtocol.Factory();
      inputProtoFactory = new TCompactProtocol.Factory(maxMessageSize, maxMessageSize);
    } else {
      protocolFactory = new TBinaryProtocol.Factory();
      inputProtoFactory = new TBinaryProtocol.Factory(true, true, maxMessageSize, maxMessageSize);
    }
    
    msHost = MetastoreConf.getVar(conf, ConfVars.THRIFT_BIND_HOST);
    if (msHost != null && !msHost.trim().isEmpty()) {
      LOG.info("Binding host " + msHost + " for metastore server");
    }
    
    IHMSHandler handler = HMSHandlerProxyFactory.getProxy(conf, baseHandler, false);
    TServerSocket serverSocket;
    if (useSasl) {
      processor = saslServer.wrapProcessor(
          new ThriftHiveMetastore.Processor<>(handler));
      LOG.info("Starting DB backed MetaStore Server in Secure Mode");
    } else {
      // we are in unsecure mode.
      if (MetastoreConf.getBoolVar(conf, ConfVars.EXECUTE_SET_UGI)) {
        processor = new TUGIBasedProcessor<>(handler);
        LOG.info("Starting DB backed MetaStore Server with SetUGI enabled");
      } else {
        processor = new TSetIpAddressProcessor<>(handler);
        LOG.info("Starting DB backed MetaStore Server");
      }
    }
    
    if (!useSSL) {
      serverSocket = SecurityUtils.getServerSocket(msHost, port);
    } else {
      String keyStorePath = MetastoreConf.getVar(conf, ConfVars.SSL_KEYSTORE_PATH).trim();
      if (keyStorePath.isEmpty()) {
        throw new IllegalArgumentException(ConfVars.SSL_KEYSTORE_PATH.toString()
            + " Not configured for SSL connection");
      }
      String keyStorePassword =
          MetastoreConf.getPassword(conf, MetastoreConf.ConfVars.SSL_KEYSTORE_PASSWORD);
      String keyStoreType =
          MetastoreConf.getVar(conf, ConfVars.SSL_KEYSTORE_TYPE).trim();
      String keyStoreAlgorithm =
          MetastoreConf.getVar(conf, ConfVars.SSL_KEYMANAGERFACTORY_ALGORITHM).trim();
      // enable SSL support for HMS
      List<String> sslVersionBlacklist = new ArrayList<>();
      for (String sslVersion : MetastoreConf.getVar(conf, ConfVars.SSL_PROTOCOL_BLACKLIST).split(",")) {
        sslVersionBlacklist.add(sslVersion);
      }

      serverSocket = SecurityUtils.getServerSSLSocket(msHost, port, keyStorePath,
          keyStorePassword, keyStoreType, keyStoreAlgorithm, sslVersionBlacklist);
    }

    ExecutorService executorService = new ThreadPoolExecutor(minWorkerThreads, maxWorkerThreads,
        60L, TimeUnit.SECONDS, new SynchronousQueue<>(), r -> {
          Thread thread = new Thread(r);
          thread.setDaemon(true);
          thread.setName("Metastore-Handler-Pool: Thread-" + thread.getId());
          return thread;
    });

    TThreadPoolServer.Args args =
        new TThreadPoolServer.Args(new TServerSocket(serverSocket.getServerSocket()) {
          @Override
          public TSocket accept() throws TTransportException {
            TSocket ts = super.accept();
            // get the limit from the configuration for every new connection
            int maxThriftMessageSize = (int) MetastoreConf.getSizeVar(
                conf, MetastoreConf.ConfVars.THRIFT_METASTORE_CLIENT_MAX_MESSAGE_SIZE);
            HMSHandler.LOG.debug("Thrift maxMessageSize = {}", maxThriftMessageSize);
            if (maxThriftMessageSize > 0) {
              ts.getConfiguration().setMaxMessageSize(maxThriftMessageSize);
            }
            return ts;
          }
        })
        .processor(processor)
        .transportFactory(transFactory)
        .protocolFactory(protocolFactory)
        .inputProtocolFactory(inputProtoFactory)
        .executorService(executorService);

    TServer tServer = new TThreadPoolServer(args);
    TServerEventHandler tServerEventHandler = new TServerEventHandler() {
      @Override
      public void preServe() {
      }

      @Override
      public ServerContext createContext(TProtocol tProtocol, TProtocol tProtocol1) {
        Metrics.getOpenConnectionsCounter().inc();
        return null;
      }

      @Override
      public void deleteContext(ServerContext serverContext, TProtocol tProtocol, TProtocol tProtocol1) {
        Metrics.getOpenConnectionsCounter().dec();
        // If the IMetaStoreClient#close was called, HMSHandler#shutdown would have already
        // cleaned up thread local RawStore. Otherwise, do it now.
        HMSHandler.cleanupHandlerContext();
      }

      @Override
      public void processContext(ServerContext serverContext, TTransport tTransport, TTransport tTransport1) {
      }
    };

    tServer.setServerEventHandler(tServerEventHandler);
    return new ThriftServer() {
      @Override
      public void start() throws Throwable {
        HMSHandler.LOG.info("Started the new metaserver on port [{}]...", port);
        HMSHandler.LOG.info("Options.minWorkerThreads = {}", minWorkerThreads);
        HMSHandler.LOG.info("Options.maxWorkerThreads = {}", maxWorkerThreads);
        HMSHandler.LOG.info("TCP keepalive = {}", tcpKeepAlive);
        HMSHandler.LOG.info("Enable SSL = {}", useSSL);
        tServer.serve();
      }

      @Override
      public boolean isRunning() {
        return tServer != null && tServer.isServing();
      }

      @Override
      public IHMSHandler getHandler() {
        return handler;
      }
    };
  }
Start the metastore store.
@param port
@param bridge
@param conf
@throws Throwable
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java