/**
 * 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.thrift;

import static org.apache.hadoop.hbase.thrift.Constants.BACKLOG_CONF_DEAFULT;
import static org.apache.hadoop.hbase.thrift.Constants.BACKLOG_CONF_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.BIND_CONF_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.BIND_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.COMPACT_CONF_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.COMPACT_CONF_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.COMPACT_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.DEFAULT_BIND_ADDR;
import static org.apache.hadoop.hbase.thrift.Constants.DEFAULT_HTTP_MAX_HEADER_SIZE;
import static org.apache.hadoop.hbase.thrift.Constants.DEFAULT_LISTEN_PORT;
import static org.apache.hadoop.hbase.thrift.Constants.FRAMED_CONF_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.FRAMED_CONF_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.FRAMED_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.HTTP_MAX_THREADS_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.HTTP_MAX_THREADS_KEY_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.HTTP_MIN_THREADS_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.HTTP_MIN_THREADS_KEY_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.INFOPORT_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.KEEP_ALIVE_SEC_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.MAX_FRAME_SIZE_CONF_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.MAX_FRAME_SIZE_CONF_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.MAX_QUEUE_SIZE_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.MAX_WORKERS_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.MIN_WORKERS_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.PORT_CONF_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.PORT_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.READ_TIMEOUT_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.SELECTOR_NUM_OPTION;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_DNS_INTERFACE_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_DNS_NAMESERVER_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_FILTERS;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_HTTP_ALLOW_OPTIONS_METHOD;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_HTTP_ALLOW_OPTIONS_METHOD_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_INFO_SERVER_BINDING_ADDRESS;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_INFO_SERVER_BINDING_ADDRESS_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_INFO_SERVER_PORT;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_INFO_SERVER_PORT_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_KERBEROS_PRINCIPAL_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_KEYTAB_FILE_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_QOP_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SELECTOR_NUM;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SPNEGO_KEYTAB_FILE_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SPNEGO_PRINCIPAL_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_ENABLED_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_EXCLUDE_CIPHER_SUITES_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_EXCLUDE_PROTOCOLS_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_INCLUDE_CIPHER_SUITES_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_INCLUDE_PROTOCOLS_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_PASSWORD_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_STORE_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SUPPORT_PROXYUSER_KEY;
import static org.apache.hadoop.hbase.thrift.Constants.USE_HTTP_CONF_KEY;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.security.auth.callback.Callback;
import javax.security.auth.callback.UnsupportedCallbackException;
import javax.security.sasl.AuthorizeCallback;
import javax.security.sasl.SaslServer;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HBaseInterfaceAudience;
import org.apache.hadoop.hbase.filter.ParseFilter;
import org.apache.hadoop.hbase.http.HttpServerUtil;
import org.apache.hadoop.hbase.http.InfoServer;
import org.apache.hadoop.hbase.security.SaslUtil;
import org.apache.hadoop.hbase.security.SecurityUtil;
import org.apache.hadoop.hbase.security.UserProvider;
import org.apache.hadoop.hbase.thrift.generated.Hbase;
import org.apache.hadoop.hbase.util.DNS;
import org.apache.hadoop.hbase.util.JvmPauseMonitor;
import org.apache.hadoop.hbase.util.Strings;
import org.apache.hadoop.hbase.util.VersionInfo;
import org.apache.hadoop.security.SaslRpcServer;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.ProxyUsers;
import org.apache.hadoop.util.Shell.ExitCodeException;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TNonblockingServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TServlet;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TNonblockingServerTransport;
import org.apache.thrift.transport.TSaslServerTransport;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TServerTransport;
import org.apache.thrift.transport.TTransportFactory;
import org.apache.yetus.audience.InterfaceAudience;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.HttpConnectionFactory;
import org.eclipse.jetty.server.SecureRequestCustomizer;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.SslConnectionFactory;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.hbase.thirdparty.com.google.common.base.Joiner;
import org.apache.hbase.thirdparty.com.google.common.base.Splitter;
import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLineParser;
import org.apache.hbase.thirdparty.org.apache.commons.cli.DefaultParser;
import org.apache.hbase.thirdparty.org.apache.commons.cli.HelpFormatter;
import org.apache.hbase.thirdparty.org.apache.commons.cli.Options;

/**
 * ThriftServer- this class starts up a Thrift server which implements the
 * Hbase API specified in the Hbase.thrift IDL file. The server runs in an
 * independent process.
 */
@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
public class ThriftServer  extends Configured implements Tool {

  private static final Logger LOG = LoggerFactory.getLogger(ThriftServer.class);



  protected Configuration conf;

  protected InfoServer infoServer;

  protected TProcessor processor;

  protected ThriftMetrics metrics;
  protected HBaseServiceHandler hbaseServiceHandler;
  protected UserGroupInformation serviceUGI;
  protected UserGroupInformation httpUGI;
  protected boolean httpEnabled;

  protected SaslUtil.QualityOfProtection qop;
  protected String host;
  protected int listenPort;


  protected boolean securityEnabled;
  protected boolean doAsEnabled;

  protected JvmPauseMonitor pauseMonitor;

  protected volatile TServer tserver;
  protected volatile Server httpServer;


  //
  // Main program and support routines
  //

  public ThriftServer(Configuration conf) {
    this.conf = HBaseConfiguration.create(conf);
  }

  protected ThriftMetrics createThriftMetrics(Configuration conf) {
    return new ThriftMetrics(conf, ThriftMetrics.ThriftServerType.ONE);
  }

  protected void setupParamters() throws IOException {
    // login the server principal (if using secure Hadoop)
    UserProvider userProvider = UserProvider.instantiate(conf);
    securityEnabled = userProvider.isHadoopSecurityEnabled()
        && userProvider.isHBaseSecurityEnabled();
    if (securityEnabled) {
      host = Strings.domainNamePointerToHostName(DNS.getDefaultHost(
          conf.get(THRIFT_DNS_INTERFACE_KEY, "default"),
          conf.get(THRIFT_DNS_NAMESERVER_KEY, "default")));
      userProvider.login(THRIFT_KEYTAB_FILE_KEY, THRIFT_KERBEROS_PRINCIPAL_KEY, host);

      // Setup the SPNEGO user for HTTP if configured
      String spnegoPrincipal = getSpengoPrincipal(conf, host);
      String spnegoKeytab = getSpnegoKeytab(conf);
      UserGroupInformation.setConfiguration(conf);
      // login the SPNEGO principal using UGI to avoid polluting the login user
      this.httpUGI = UserGroupInformation.loginUserFromKeytabAndReturnUGI(spnegoPrincipal,
        spnegoKeytab);
    }
    this.serviceUGI = userProvider.getCurrent().getUGI();
    if (httpUGI == null) {
      this.httpUGI = serviceUGI;
    }

    this.listenPort = conf.getInt(PORT_CONF_KEY, DEFAULT_LISTEN_PORT);
    this.metrics = createThriftMetrics(conf);
    this.pauseMonitor = new JvmPauseMonitor(conf, this.metrics.getSource());
    this.hbaseServiceHandler = createHandler(conf, userProvider);
    this.hbaseServiceHandler.initMetrics(metrics);
    this.processor = createProcessor();

    httpEnabled = conf.getBoolean(USE_HTTP_CONF_KEY, false);
    doAsEnabled = conf.getBoolean(THRIFT_SUPPORT_PROXYUSER_KEY, false);
    if (doAsEnabled && !httpEnabled) {
      LOG.warn("Fail to enable the doAs feature. " + USE_HTTP_CONF_KEY + " is not configured");
    }

    String strQop = conf.get(THRIFT_QOP_KEY);
    if (strQop != null) {
      this.qop = SaslUtil.getQop(strQop);
    }
    if (qop != null) {
      if (qop != SaslUtil.QualityOfProtection.AUTHENTICATION &&
          qop != SaslUtil.QualityOfProtection.INTEGRITY &&
          qop != SaslUtil.QualityOfProtection.PRIVACY) {
        throw new IOException(String.format("Invalid %s: It must be one of %s, %s, or %s.",
            THRIFT_QOP_KEY,
            SaslUtil.QualityOfProtection.AUTHENTICATION.name(),
            SaslUtil.QualityOfProtection.INTEGRITY.name(),
            SaslUtil.QualityOfProtection.PRIVACY.name()));
      }
      checkHttpSecurity(qop, conf);
      if (!securityEnabled) {
        throw new IOException("Thrift server must run in secure mode to support authentication");
      }
    }
    registerFilters(conf);
    pauseMonitor.start();
  }

  private String getSpengoPrincipal(Configuration conf, String host) throws IOException {
    String principal = conf.get(THRIFT_SPNEGO_PRINCIPAL_KEY);
    if (principal == null) {
      // We cannot use the Hadoop configuration deprecation handling here since
      // the THRIFT_KERBEROS_PRINCIPAL_KEY config is still valid for regular Kerberos
      // communication. The preference should be to use the THRIFT_SPNEGO_PRINCIPAL_KEY
      // config so that THRIFT_KERBEROS_PRINCIPAL_KEY doesn't control both backend
      // Kerberos principal and SPNEGO principal.
      LOG.info("Using deprecated {} config for SPNEGO principal. Use {} instead.",
        THRIFT_KERBEROS_PRINCIPAL_KEY, THRIFT_SPNEGO_PRINCIPAL_KEY);
      principal = conf.get(THRIFT_KERBEROS_PRINCIPAL_KEY);
    }
    // Handle _HOST in principal value
    return org.apache.hadoop.security.SecurityUtil.getServerPrincipal(principal, host);
  }

  private String getSpnegoKeytab(Configuration conf) {
    String keytab = conf.get(THRIFT_SPNEGO_KEYTAB_FILE_KEY);
    if (keytab == null) {
      // We cannot use the Hadoop configuration deprecation handling here since
      // the THRIFT_KEYTAB_FILE_KEY config is still valid for regular Kerberos
      // communication. The preference should be to use the THRIFT_SPNEGO_KEYTAB_FILE_KEY
      // config so that THRIFT_KEYTAB_FILE_KEY doesn't control both backend
      // Kerberos keytab and SPNEGO keytab.
      LOG.info("Using deprecated {} config for SPNEGO keytab. Use {} instead.",
        THRIFT_KEYTAB_FILE_KEY, THRIFT_SPNEGO_KEYTAB_FILE_KEY);
      keytab = conf.get(THRIFT_KEYTAB_FILE_KEY);
    }
    return keytab;
  }

  protected void startInfoServer() throws IOException {
    // Put up info server.
    int port = conf.getInt(THRIFT_INFO_SERVER_PORT , THRIFT_INFO_SERVER_PORT_DEFAULT);

    if (port >= 0) {
      conf.setLong("startcode", System.currentTimeMillis());
      String a = conf
          .get(THRIFT_INFO_SERVER_BINDING_ADDRESS, THRIFT_INFO_SERVER_BINDING_ADDRESS_DEFAULT);
      infoServer = new InfoServer("thrift", a, port, false, conf);
      infoServer.setAttribute("hbase.conf", conf);
      infoServer.start();
    }
  }

  protected void checkHttpSecurity(SaslUtil.QualityOfProtection qop, Configuration conf) {
    if (qop == SaslUtil.QualityOfProtection.PRIVACY &&
        conf.getBoolean(USE_HTTP_CONF_KEY, false) &&
        !conf.getBoolean(THRIFT_SSL_ENABLED_KEY, false)) {
      throw new IllegalArgumentException("Thrift HTTP Server's QoP is privacy, but " +
          THRIFT_SSL_ENABLED_KEY + " is false");
    }
  }

  protected HBaseServiceHandler createHandler(Configuration conf, UserProvider userProvider)
      throws IOException {
    return new ThriftHBaseServiceHandler(conf, userProvider);
  }

  protected TProcessor createProcessor() {
    return new Hbase.Processor<>(
        HbaseHandlerMetricsProxy.newInstance((Hbase.Iface) hbaseServiceHandler, metrics, conf));
  }

  /**
   * the thrift server, not null means the server is started, for test only
   * @return the tServer
   */
  @VisibleForTesting
  public TServer getTserver() {
    return tserver;
  }

  /**
   * the Jetty server, not null means the HTTP server is started, for test only
   * @return the http server
   */
  @VisibleForTesting
  public Server getHttpServer() {
    return httpServer;
  }

  protected void printUsageAndExit(Options options, int exitCode)
      throws ExitCodeException {
    HelpFormatter formatter = new HelpFormatter();
    formatter.printHelp("Thrift", null, options,
        "To start the Thrift server run 'hbase-daemon.sh start thrift' or " +
            "'hbase thrift'\n" +
            "To shutdown the thrift server run 'hbase-daemon.sh stop " +
            "thrift' or send a kill signal to the thrift server pid",
        true);
    throw new ExitCodeException(exitCode, "");
  }

  /**
   * Create a Servlet for the http server
   * @param protocolFactory protocolFactory
   * @return the servlet
   */
  protected TServlet createTServlet(TProtocolFactory protocolFactory) {
    return new ThriftHttpServlet(processor, protocolFactory, serviceUGI, httpUGI,
        hbaseServiceHandler, securityEnabled, doAsEnabled);
  }

  /**
   * Setup an HTTP Server using Jetty to serve calls from THttpClient
   *
   * @throws IOException IOException
   */
  protected void setupHTTPServer() throws IOException {
    TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
    TServlet thriftHttpServlet = createTServlet(protocolFactory);

    // Set the default max thread number to 100 to limit
    // the number of concurrent requests so that Thrfit HTTP server doesn't OOM easily.
    // Jetty set the default max thread number to 250, if we don't set it.
    //
    // Our default min thread number 2 is the same as that used by Jetty.
    int minThreads = conf.getInt(HTTP_MIN_THREADS_KEY,
        conf.getInt(TBoundedThreadPoolServer.MIN_WORKER_THREADS_CONF_KEY,
            HTTP_MIN_THREADS_KEY_DEFAULT));
    int maxThreads = conf.getInt(HTTP_MAX_THREADS_KEY,
        conf.getInt(TBoundedThreadPoolServer.MAX_WORKER_THREADS_CONF_KEY,
            HTTP_MAX_THREADS_KEY_DEFAULT));
    QueuedThreadPool threadPool = new QueuedThreadPool(maxThreads);
    threadPool.setMinThreads(minThreads);
    httpServer = new Server(threadPool);

    // Context handler
    ServletContextHandler ctxHandler = new ServletContextHandler(httpServer, "/",
        ServletContextHandler.SESSIONS);
    ctxHandler.addServlet(new ServletHolder(thriftHttpServlet), "/*");
    HttpServerUtil.constrainHttpMethods(ctxHandler,
        conf.getBoolean(THRIFT_HTTP_ALLOW_OPTIONS_METHOD,
            THRIFT_HTTP_ALLOW_OPTIONS_METHOD_DEFAULT));

    // set up Jetty and run the embedded server
    HttpConfiguration httpConfig = new HttpConfiguration();
    httpConfig.setSecureScheme("https");
    httpConfig.setSecurePort(listenPort);
    httpConfig.setHeaderCacheSize(DEFAULT_HTTP_MAX_HEADER_SIZE);
    httpConfig.setRequestHeaderSize(DEFAULT_HTTP_MAX_HEADER_SIZE);
    httpConfig.setResponseHeaderSize(DEFAULT_HTTP_MAX_HEADER_SIZE);
    httpConfig.setSendServerVersion(false);
    httpConfig.setSendDateHeader(false);

    ServerConnector serverConnector;
    if(conf.getBoolean(THRIFT_SSL_ENABLED_KEY, false)) {
      HttpConfiguration httpsConfig = new HttpConfiguration(httpConfig);
      httpsConfig.addCustomizer(new SecureRequestCustomizer());

      SslContextFactory sslCtxFactory = new SslContextFactory();
      String keystore = conf.get(THRIFT_SSL_KEYSTORE_STORE_KEY);
      String password = HBaseConfiguration.getPassword(conf,
          THRIFT_SSL_KEYSTORE_PASSWORD_KEY, null);
      String keyPassword = HBaseConfiguration.getPassword(conf,
          THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY, password);
      sslCtxFactory.setKeyStorePath(keystore);
      sslCtxFactory.setKeyStorePassword(password);
      sslCtxFactory.setKeyManagerPassword(keyPassword);

      String[] excludeCiphers = conf.getStrings(
          THRIFT_SSL_EXCLUDE_CIPHER_SUITES_KEY, ArrayUtils.EMPTY_STRING_ARRAY);
      if (excludeCiphers.length != 0) {
        sslCtxFactory.setExcludeCipherSuites(excludeCiphers);
      }
      String[] includeCiphers = conf.getStrings(
          THRIFT_SSL_INCLUDE_CIPHER_SUITES_KEY, ArrayUtils.EMPTY_STRING_ARRAY);
      if (includeCiphers.length != 0) {
        sslCtxFactory.setIncludeCipherSuites(includeCiphers);
      }

      // Disable SSLv3 by default due to "Poodle" Vulnerability - CVE-2014-3566
      String[] excludeProtocols = conf.getStrings(
          THRIFT_SSL_EXCLUDE_PROTOCOLS_KEY, "SSLv3");
      if (excludeProtocols.length != 0) {
        sslCtxFactory.setExcludeProtocols(excludeProtocols);
      }
      String[] includeProtocols = conf.getStrings(
          THRIFT_SSL_INCLUDE_PROTOCOLS_KEY, ArrayUtils.EMPTY_STRING_ARRAY);
      if (includeProtocols.length != 0) {
        sslCtxFactory.setIncludeProtocols(includeProtocols);
      }

      serverConnector = new ServerConnector(httpServer,
          new SslConnectionFactory(sslCtxFactory, HttpVersion.HTTP_1_1.toString()),
          new HttpConnectionFactory(httpsConfig));
    } else {
      serverConnector = new ServerConnector(httpServer, new HttpConnectionFactory(httpConfig));
    }
    serverConnector.setPort(listenPort);
    serverConnector.setHost(getBindAddress(conf).getHostAddress());
    httpServer.addConnector(serverConnector);
    httpServer.setStopAtShutdown(true);

    if (doAsEnabled) {
      ProxyUsers.refreshSuperUserGroupsConfiguration(conf);
    }
    LOG.info("Starting Thrift HTTP Server on {}", Integer.toString(listenPort));
  }

  /**
   * Setting up the thrift TServer
   */
  protected void setupServer() throws Exception {
    // Construct correct ProtocolFactory
    TProtocolFactory protocolFactory = getProtocolFactory();

    ImplType implType = ImplType.getServerImpl(conf);
    TProcessor processorToUse = processor;

    // Construct correct TransportFactory
    TTransportFactory transportFactory;
    if (conf.getBoolean(FRAMED_CONF_KEY, FRAMED_CONF_DEFAULT) || implType.isAlwaysFramed) {
      if (qop != null) {
        throw new RuntimeException("Thrift server authentication"
            + " doesn't work with framed transport yet");
      }
      transportFactory = new TFramedTransport.Factory(
          conf.getInt(MAX_FRAME_SIZE_CONF_KEY, MAX_FRAME_SIZE_CONF_DEFAULT) * 1024 * 1024);
      LOG.debug("Using framed transport");
    } else if (qop == null) {
      transportFactory = new TTransportFactory();
    } else {
      // Extract the name from the principal
      String thriftKerberosPrincipal = conf.get(THRIFT_KERBEROS_PRINCIPAL_KEY);
      if (thriftKerberosPrincipal == null) {
        throw new IllegalArgumentException(THRIFT_KERBEROS_PRINCIPAL_KEY + " cannot be null");
      }
      String name = SecurityUtil.getUserFromPrincipal(thriftKerberosPrincipal);
      Map<String, String> saslProperties = SaslUtil.initSaslProperties(qop.name());
      TSaslServerTransport.Factory saslFactory = new TSaslServerTransport.Factory();
      saslFactory.addServerDefinition("GSSAPI", name, host, saslProperties,
          new SaslRpcServer.SaslGssCallbackHandler() {
            @Override
            public void handle(Callback[] callbacks)
                throws UnsupportedCallbackException {
              AuthorizeCallback ac = null;
              for (Callback callback : callbacks) {
                if (callback instanceof AuthorizeCallback) {
                  ac = (AuthorizeCallback) callback;
                } else {
                  throw new UnsupportedCallbackException(callback,
                      "Unrecognized SASL GSSAPI Callback");
                }
              }
              if (ac != null) {
                String authid = ac.getAuthenticationID();
                String authzid = ac.getAuthorizationID();
                if (!authid.equals(authzid)) {
                  ac.setAuthorized(false);
                } else {
                  ac.setAuthorized(true);
                  String userName = SecurityUtil.getUserFromPrincipal(authzid);
                  LOG.info("Effective user: {}", userName);
                  ac.setAuthorizedID(userName);
                }
              }
            }
          });
      transportFactory = saslFactory;

      // Create a processor wrapper, to get the caller
      processorToUse = (inProt, outProt) -> {
        TSaslServerTransport saslServerTransport =
            (TSaslServerTransport)inProt.getTransport();
        SaslServer saslServer = saslServerTransport.getSaslServer();
        String principal = saslServer.getAuthorizationID();
        hbaseServiceHandler.setEffectiveUser(principal);
        processor.process(inProt, outProt);
      };
    }

    if (conf.get(BIND_CONF_KEY) != null && !implType.canSpecifyBindIP) {
      LOG.error("Server types {} don't support IP address binding at the moment. See " +
              "https://issues.apache.org/jira/browse/HBASE-2155 for details.",
          Joiner.on(", ").join(ImplType.serversThatCannotSpecifyBindIP()));
      throw new RuntimeException("-" + BIND_CONF_KEY + " not supported with " + implType);
    }

    InetSocketAddress inetSocketAddress = new InetSocketAddress(getBindAddress(conf), listenPort);
    if (implType == ImplType.HS_HA || implType == ImplType.NONBLOCKING ||
        implType == ImplType.THREADED_SELECTOR) {
      TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
      if (implType == ImplType.NONBLOCKING) {
        tserver = getTNonBlockingServer(serverTransport, protocolFactory, processorToUse,
            transportFactory, inetSocketAddress);
      } else if (implType == ImplType.HS_HA) {
        tserver = getTHsHaServer(serverTransport, protocolFactory, processorToUse, transportFactory,
            inetSocketAddress);
      } else { // THREADED_SELECTOR
        tserver = getTThreadedSelectorServer(serverTransport, protocolFactory, processorToUse,
            transportFactory, inetSocketAddress);
      }
      LOG.info("starting HBase {} server on {}", implType.simpleClassName(),
          Integer.toString(listenPort));
    } else if (implType == ImplType.THREAD_POOL) {
      this.tserver = getTThreadPoolServer(protocolFactory, processorToUse, transportFactory,
          inetSocketAddress);
    } else {
      throw new AssertionError("Unsupported Thrift server implementation: " +
          implType.simpleClassName());
    }

    // A sanity check that we instantiated the right type of server.
    if (tserver.getClass() != implType.serverClass) {
      throw new AssertionError("Expected to create Thrift server class " +
          implType.serverClass.getName() + " but got " +
          tserver.getClass().getName());
    }
  }

  protected TServer getTNonBlockingServer(TNonblockingServerTransport serverTransport,
      TProtocolFactory protocolFactory, TProcessor processor, TTransportFactory transportFactory,
      InetSocketAddress inetSocketAddress) {
    LOG.info("starting HBase Nonblocking Thrift server on " + inetSocketAddress.toString());
    TNonblockingServer.Args serverArgs = new TNonblockingServer.Args(serverTransport);
    serverArgs.processor(processor);
    serverArgs.transportFactory(transportFactory);
    serverArgs.protocolFactory(protocolFactory);
    return new TNonblockingServer(serverArgs);
  }

  protected TServer getTHsHaServer(TNonblockingServerTransport serverTransport,
      TProtocolFactory protocolFactory, TProcessor processor, TTransportFactory transportFactory,
      InetSocketAddress inetSocketAddress) {
    LOG.info("starting HBase HsHA Thrift server on " + inetSocketAddress.toString());
    THsHaServer.Args serverArgs = new THsHaServer.Args(serverTransport);
    int queueSize = conf.getInt(TBoundedThreadPoolServer.MAX_QUEUED_REQUESTS_CONF_KEY,
        TBoundedThreadPoolServer.DEFAULT_MAX_QUEUED_REQUESTS);
    CallQueue callQueue = new CallQueue(new LinkedBlockingQueue<>(queueSize), metrics);
    int workerThread = conf.getInt(TBoundedThreadPoolServer.MAX_WORKER_THREADS_CONF_KEY,
        serverArgs.getMaxWorkerThreads());
    ExecutorService executorService = createExecutor(
        callQueue, workerThread, workerThread);
    serverArgs.executorService(executorService).processor(processor)
        .transportFactory(transportFactory).protocolFactory(protocolFactory);
    return new THsHaServer(serverArgs);
  }

  protected TServer getTThreadedSelectorServer(TNonblockingServerTransport serverTransport,
      TProtocolFactory protocolFactory, TProcessor processor, TTransportFactory transportFactory,
      InetSocketAddress inetSocketAddress) {
    LOG.info("starting HBase ThreadedSelector Thrift server on " + inetSocketAddress.toString());
    TThreadedSelectorServer.Args serverArgs =
        new HThreadedSelectorServerArgs(serverTransport, conf);
    int queueSize = conf.getInt(TBoundedThreadPoolServer.MAX_QUEUED_REQUESTS_CONF_KEY,
        TBoundedThreadPoolServer.DEFAULT_MAX_QUEUED_REQUESTS);
    CallQueue callQueue = new CallQueue(new LinkedBlockingQueue<>(queueSize), metrics);
    int workerThreads = conf.getInt(TBoundedThreadPoolServer.MAX_WORKER_THREADS_CONF_KEY,
        serverArgs.getWorkerThreads());
    int selectorThreads = conf.getInt(THRIFT_SELECTOR_NUM, serverArgs.getSelectorThreads());
    serverArgs.selectorThreads(selectorThreads);
    ExecutorService executorService = createExecutor(
        callQueue, workerThreads, workerThreads);
    serverArgs.executorService(executorService).processor(processor)
        .transportFactory(transportFactory).protocolFactory(protocolFactory);
    return new TThreadedSelectorServer(serverArgs);
  }

  protected TServer getTThreadPoolServer(TProtocolFactory protocolFactory, TProcessor processor,
      TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) throws Exception {
    LOG.info("starting HBase ThreadPool Thrift server on " + inetSocketAddress.toString());
    // Thrift's implementation uses '0' as a placeholder for 'use the default.'
    int backlog = conf.getInt(BACKLOG_CONF_KEY, BACKLOG_CONF_DEAFULT);
    int readTimeout = conf.getInt(THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY,
        THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT);
    TServerTransport serverTransport = new TServerSocket(
        new TServerSocket.ServerSocketTransportArgs().
            bindAddr(inetSocketAddress).backlog(backlog).
            clientTimeout(readTimeout));

    TBoundedThreadPoolServer.Args serverArgs =
        new TBoundedThreadPoolServer.Args(serverTransport, conf);
    serverArgs.processor(processor).transportFactory(transportFactory)
        .protocolFactory(protocolFactory);
    return new TBoundedThreadPoolServer(serverArgs, metrics);
  }

  protected TProtocolFactory getProtocolFactory() {
    TProtocolFactory protocolFactory;

    if (conf.getBoolean(COMPACT_CONF_KEY, COMPACT_CONF_DEFAULT)) {
      LOG.debug("Using compact protocol");
      protocolFactory = new TCompactProtocol.Factory();
    } else {
      LOG.debug("Using binary protocol");
      protocolFactory = new TBinaryProtocol.Factory();
    }

    return protocolFactory;
  }

  protected ExecutorService createExecutor(BlockingQueue<Runnable> callQueue,
      int minWorkers, int maxWorkers) {
    ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
    tfb.setDaemon(true);
    tfb.setNameFormat("thrift-worker-%d");
    ThreadPoolExecutor threadPool = new THBaseThreadPoolExecutor(minWorkers, maxWorkers,
        Long.MAX_VALUE, TimeUnit.SECONDS, callQueue, tfb.build(), metrics);
    threadPool.allowCoreThreadTimeOut(true);
    return threadPool;
  }

  protected InetAddress getBindAddress(Configuration conf)
      throws UnknownHostException {
    String bindAddressStr = conf.get(BIND_CONF_KEY, DEFAULT_BIND_ADDR);
    return InetAddress.getByName(bindAddressStr);
  }


  public static void registerFilters(Configuration conf) {
    String[] filters = conf.getStrings(THRIFT_FILTERS);
    Splitter splitter = Splitter.on(':');
    if(filters != null) {
      for(String filterClass: filters) {
        List<String> filterPart = splitter.splitToList(filterClass);
        if(filterPart.size() != 2) {
          LOG.warn("Invalid filter specification " + filterClass + " - skipping");
        } else {
          ParseFilter.registerFilter(filterPart.get(0), filterPart.get(1));
        }
      }
    }
  }

  /**
   * Add options to command lines
   * @param options options
   */
  protected void addOptions(Options options) {
    options.addOption("b", BIND_OPTION, true, "Address to bind " +
        "the Thrift server to. [default: " + DEFAULT_BIND_ADDR + "]");
    options.addOption("p", PORT_OPTION, true, "Port to bind to [default: " +
        DEFAULT_LISTEN_PORT + "]");
    options.addOption("f", FRAMED_OPTION, false, "Use framed transport");
    options.addOption("c", COMPACT_OPTION, false, "Use the compact protocol");
    options.addOption("h", "help", false, "Print help information");
    options.addOption("s", SELECTOR_NUM_OPTION, true, "How many selector threads to use.");
    options.addOption(null, INFOPORT_OPTION, true, "Port for web UI");

    options.addOption("m", MIN_WORKERS_OPTION, true,
        "The minimum number of worker threads for " +
            ImplType.THREAD_POOL.simpleClassName());

    options.addOption("w", MAX_WORKERS_OPTION, true,
        "The maximum number of worker threads for " +
            ImplType.THREAD_POOL.simpleClassName());

    options.addOption("q", MAX_QUEUE_SIZE_OPTION, true,
        "The maximum number of queued requests in " +
            ImplType.THREAD_POOL.simpleClassName());

    options.addOption("k", KEEP_ALIVE_SEC_OPTION, true,
        "The amount of time in secods to keep a thread alive when idle in " +
            ImplType.THREAD_POOL.simpleClassName());

    options.addOption("t", READ_TIMEOUT_OPTION, true,
        "Amount of time in milliseconds before a server thread will timeout " +
            "waiting for client to send data on a connected socket. Currently, " +
            "only applies to TBoundedThreadPoolServer");

    options.addOptionGroup(ImplType.createOptionGroup());
  }

  protected void parseCommandLine(CommandLine cmd, Options options) throws ExitCodeException {
    // Get port to bind to
    try {
      if (cmd.hasOption(PORT_OPTION)) {
        int listenPort = Integer.parseInt(cmd.getOptionValue(PORT_OPTION));
        conf.setInt(PORT_CONF_KEY, listenPort);
      }
    } catch (NumberFormatException e) {
      LOG.error("Could not parse the value provided for the port option", e);
      printUsageAndExit(options, -1);
    }
    // check for user-defined info server port setting, if so override the conf
    try {
      if (cmd.hasOption(INFOPORT_OPTION)) {
        String val = cmd.getOptionValue(INFOPORT_OPTION);
        conf.setInt(THRIFT_INFO_SERVER_PORT, Integer.parseInt(val));
        LOG.debug("Web UI port set to " + val);
      }
    } catch (NumberFormatException e) {
      LOG.error("Could not parse the value provided for the " + INFOPORT_OPTION +
          " option", e);
      printUsageAndExit(options, -1);
    }
    // Make optional changes to the configuration based on command-line options
    optionToConf(cmd, MIN_WORKERS_OPTION,
        conf, TBoundedThreadPoolServer.MIN_WORKER_THREADS_CONF_KEY);
    optionToConf(cmd, MAX_WORKERS_OPTION,
        conf, TBoundedThreadPoolServer.MAX_WORKER_THREADS_CONF_KEY);
    optionToConf(cmd, MAX_QUEUE_SIZE_OPTION,
        conf, TBoundedThreadPoolServer.MAX_QUEUED_REQUESTS_CONF_KEY);
    optionToConf(cmd, KEEP_ALIVE_SEC_OPTION,
        conf, TBoundedThreadPoolServer.THREAD_KEEP_ALIVE_TIME_SEC_CONF_KEY);
    optionToConf(cmd, READ_TIMEOUT_OPTION, conf, THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY);
    optionToConf(cmd, SELECTOR_NUM_OPTION, conf, THRIFT_SELECTOR_NUM);

    // Set general thrift server options
    boolean compact = cmd.hasOption(COMPACT_OPTION) ||
        conf.getBoolean(COMPACT_CONF_KEY, false);
    conf.setBoolean(COMPACT_CONF_KEY, compact);
    boolean framed = cmd.hasOption(FRAMED_OPTION) ||
        conf.getBoolean(FRAMED_CONF_KEY, false);
    conf.setBoolean(FRAMED_CONF_KEY, framed);

    optionToConf(cmd, BIND_OPTION, conf, BIND_CONF_KEY);


    ImplType.setServerImpl(cmd, conf);
  }

  /**
   * Parse the command line options to set parameters the conf.
   */
  protected void processOptions(final String[] args) throws Exception {
    if (args == null || args.length == 0) {
      return;
    }
    Options options = new Options();
    addOptions(options);

    CommandLineParser parser = new DefaultParser();
    CommandLine cmd = parser.parse(options, args);

    if (cmd.hasOption("help")) {
      printUsageAndExit(options, 1);
    }
    parseCommandLine(cmd, options);
  }

  public void stop() {
    if (this.infoServer != null) {
      LOG.info("Stopping infoServer");
      try {
        this.infoServer.stop();
      } catch (Exception ex) {
        LOG.error("Failed to stop infoServer", ex);
      }
    }
    if (pauseMonitor != null) {
      pauseMonitor.stop();
    }
    if (tserver != null) {
      tserver.stop();
      tserver = null;
    }
    if (httpServer != null) {
      try {
        httpServer.stop();
        httpServer = null;
      } catch (Exception e) {
        LOG.error("Problem encountered in shutting down HTTP server", e);
      }
      httpServer = null;
    }
  }

  protected static void optionToConf(CommandLine cmd, String option,
      Configuration conf, String destConfKey) {
    if (cmd.hasOption(option)) {
      String value = cmd.getOptionValue(option);
      LOG.info("Set configuration key:" + destConfKey + " value:" + value);
      conf.set(destConfKey, value);
    }
  }

  /**
   * Run without any command line arguments
   * @return exit code
   * @throws Exception exception
   */
  public int run() throws Exception {
    return run(null);
  }

  @Override
  public int run(String[] strings) throws Exception {
    processOptions(strings);
    setupParamters();
    startInfoServer();
    if (httpEnabled) {
      setupHTTPServer();
      httpServer.start();
      httpServer.join();
    } else {
      setupServer();
      tserver.serve();
    }
    return 0;
  }

  public static void main(String [] args) throws Exception {
    LOG.info("***** STARTING service '" + ThriftServer.class.getSimpleName() + "' *****");
    VersionInfo.logVersion();
    final Configuration conf = HBaseConfiguration.create();
    // for now, only time we return is on an argument error.
    final int status = ToolRunner.run(conf, new ThriftServer(conf), args);
    LOG.info("***** STOPPING service '" + ThriftServer.class.getSimpleName() + "' *****");
    System.exit(status);
  }
}
