/*
 * 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.phoenix.queryserver.server;

import org.apache.calcite.avatica.Meta;
import org.apache.calcite.avatica.remote.Driver;
import org.apache.calcite.avatica.remote.LocalService;
import org.apache.calcite.avatica.remote.Service;
import org.apache.calcite.avatica.server.AvaticaServerConfiguration;
import org.apache.calcite.avatica.server.DoAsRemoteUserCallback;
import org.apache.calcite.avatica.server.HttpServer;
import org.apache.calcite.avatica.server.HttpServer.Builder;
import org.apache.calcite.avatica.server.RemoteUserExtractor;
import org.apache.calcite.avatica.server.RemoteUserExtractionException;
import org.apache.calcite.avatica.server.HttpRequestRemoteUserExtractor;
import org.apache.calcite.avatica.server.HttpQueryStringParameterRemoteUserExtractor;
import org.apache.calcite.avatica.server.ServerCustomizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.util.Strings;
import org.apache.hadoop.net.DNS;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.ProxyUsers;
import org.apache.hadoop.security.authorize.AuthorizationException;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.phoenix.loadbalancer.service.LoadBalanceZookeeperConf;
import org.apache.phoenix.queryserver.QueryServerOptions;
import org.apache.phoenix.queryserver.QueryServerProperties;
import org.apache.phoenix.queryserver.register.Registry;
import org.apache.phoenix.util.InstanceResolver;
import org.apache.phoenix.util.SimpleLRUCache;
import org.eclipse.jetty.server.Server;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.lang.management.RuntimeMXBean;
import java.net.InetAddress;
import java.security.PrivilegedExceptionAction;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.ServiceConfigurationError;
import java.util.ServiceLoader;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import javax.servlet.http.HttpServletRequest;

/**
 * A query server for Phoenix over Calcite's Avatica.
 */
public final class QueryServer extends Configured implements Tool, Runnable {

  protected static final Logger LOG = LoggerFactory.getLogger(QueryServer.class);

  private final String[] argv;
  private final CountDownLatch runningLatch = new CountDownLatch(1);
  private HttpServer server = null;
  private int retCode = 0;
  private Throwable t = null;
  private Registry registry;

  /**
   * Log information about the currently running JVM.
   */
  public static void logJVMInfo() {
    // Print out vm stats before starting up.
    RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
    if (runtime != null) {
      LOG.info("vmName=" + runtime.getVmName() + ", vmVendor=" +
              runtime.getVmVendor() + ", vmVersion=" + runtime.getVmVersion());
      LOG.info("vmInputArguments=" + runtime.getInputArguments());
    }
  }

  /**
   * Logs information about the currently running JVM process including
   * the environment variables. Logging of env vars can be disabled by
   * setting {@code "phoenix.envvars.logging.disabled"} to {@code "true"}.
   * <p>If enabled, you can also exclude environment variables containing
   * certain substrings by setting {@code "phoenix.envvars.logging.skipwords"}
   * to comma separated list of such substrings.
   */
  public static void logProcessInfo(Configuration conf) {
    // log environment variables unless asked not to
    if (conf == null || !conf.getBoolean(QueryServerProperties.QUERY_SERVER_ENV_LOGGING_ATTRIB, false)) {
      Set<String> skipWords = new HashSet<String>(
          QueryServerOptions.DEFAULT_QUERY_SERVER_SKIP_WORDS);
      if (conf != null) {
        String[] confSkipWords = conf.getStrings(
            QueryServerProperties.QUERY_SERVER_ENV_LOGGING_SKIPWORDS_ATTRIB);
        if (confSkipWords != null) {
          skipWords.addAll(Arrays.asList(confSkipWords));
        }
      }

      nextEnv:
      for (Map.Entry<String, String> entry : System.getenv().entrySet()) {
        String key = entry.getKey().toLowerCase();
        String value = entry.getValue().toLowerCase();
        // exclude variables which may contain skip words
        for(String skipWord : skipWords) {
          if (key.contains(skipWord) || value.contains(skipWord))
            continue nextEnv;
        }
        LOG.info("env:"+entry);
      }
    }
    // and JVM info
    logJVMInfo();
  }

  /** Constructor for use from {@link org.apache.hadoop.util.ToolRunner}. */
  public QueryServer() {
    this(null, null);
  }

  /** Constructor for use as {@link java.lang.Runnable}. */
  public QueryServer(String[] argv, Configuration conf) {
    this.argv = argv;
    setConf(conf);
  }

  /**
   * @return the port number this instance is bound to, or {@code -1} if the server is not running.
   */
  //@VisibleForTesting
  public int getPort() {
    if (server == null) return -1;
    return server.getPort();
  }

  /**
   * @return the return code from running as a {@link Tool}.
   */
  //@VisibleForTesting
  public int getRetCode() {
    return retCode;
  }

  /**
   * @return the throwable from an unsuccessful run, or null otherwise.
   */
  //@VisibleForTesting
  public Throwable getThrowable() {
    return t;
  }

  /** Calling thread waits until the server is running. */
  public void awaitRunning() throws InterruptedException {
    runningLatch.await();
  }

  /** Calling thread waits until the server is running. */
  public void awaitRunning(long timeout, TimeUnit unit) throws InterruptedException {
    runningLatch.await(timeout, unit);
  }

  @Override
  public int run(String[] args) throws Exception {
    logProcessInfo(getConf());
    final boolean loadBalancerEnabled = getConf().getBoolean(QueryServerProperties.PHOENIX_QUERY_SERVER_LOADBALANCER_ENABLED,
            QueryServerOptions.DEFAULT_PHOENIX_QUERY_SERVER_LOADBALANCER_ENABLED);
    try {
      final boolean isKerberos = "kerberos".equalsIgnoreCase(getConf().get(
          QueryServerProperties.QUERY_SERVER_HBASE_SECURITY_CONF_ATTRIB));
      final boolean isHadoopKerberos = "kerberos".equalsIgnoreCase(getConf().get(
          QueryServerProperties.QUERY_SERVER_HADOOP_SECURITY_CONF_ATTRIB));
      final boolean disableSpnego = getConf().getBoolean(QueryServerProperties.QUERY_SERVER_SPNEGO_AUTH_DISABLED_ATTRIB,
              QueryServerOptions.DEFAULT_QUERY_SERVER_SPNEGO_AUTH_DISABLED);
      String hostname;
      final boolean disableLogin = getConf().getBoolean(QueryServerProperties.QUERY_SERVER_DISABLE_KERBEROS_LOGIN,
              QueryServerOptions.DEFAULT_QUERY_SERVER_DISABLE_KERBEROS_LOGIN);

      // handle secure cluster credentials
      if (isKerberos && !disableLogin) {
        if(!isHadoopKerberos) {
          LOG.error("HBase and Hadoop security config inconsistent, "
                  + QueryServerProperties.QUERY_SERVER_HBASE_SECURITY_CONF_ATTRIB
                  + " was configured as kerberos, but "
                  + QueryServerProperties.QUERY_SERVER_HADOOP_SECURITY_CONF_ATTRIB + " not!");
          return -1;
        }
        hostname = Strings.domainNamePointerToHostName(DNS.getDefaultHost(
            getConf().get(QueryServerProperties.QUERY_SERVER_DNS_INTERFACE_ATTRIB, "default"),
            getConf().get(QueryServerProperties.QUERY_SERVER_DNS_NAMESERVER_ATTRIB, "default")));
        if (LOG.isDebugEnabled()) {
          LOG.debug("Login to " + hostname + " using " + getConf().get(
              QueryServerProperties.QUERY_SERVER_KEYTAB_FILENAME_ATTRIB)
              + " and principal " + getConf().get(
                  QueryServerProperties.QUERY_SERVER_KERBEROS_PRINCIPAL_ATTRIB) + ".");
        }
        SecurityUtil.login(getConf(), QueryServerProperties.QUERY_SERVER_KEYTAB_FILENAME_ATTRIB,
            QueryServerProperties.QUERY_SERVER_KERBEROS_PRINCIPAL_ATTRIB, hostname);
        LOG.info("Kerberos login successful.");
      } else {
        hostname = InetAddress.getLocalHost().getHostName();
        LOG.info("Kerberos is off and hostname is : " + hostname);
      }

      int port = getConf().getInt(QueryServerProperties.QUERY_SERVER_HTTP_PORT_ATTRIB,
          QueryServerOptions.DEFAULT_QUERY_SERVER_HTTP_PORT);
      LOG.debug("Listening on port " + port);

      // Update proxyuser configuration for impersonation
      ProxyUsers.refreshSuperUserGroupsConfiguration(getConf());

      // Start building the Avatica HttpServer
      final HttpServer.Builder<Server>
              builder =
              HttpServer.Builder.<Server>newBuilder().withPort(port);

      UserGroupInformation ugi = getUserGroupInformation();

      AvaticaServerConfiguration avaticaServerConfiguration = null;

      // RemoteUserCallbacks and RemoteUserExtractor are part of AvaticaServerConfiguration
      // Hence they should be customizable when using QUERY_SERVER_CUSTOM_AUTH_ENABLED
      // Handlers should be customized via ServerCustomizers
      if (getConf().getBoolean(QueryServerProperties.QUERY_SERVER_CUSTOM_AUTH_ENABLED,
          QueryServerOptions.DEFAULT_QUERY_SERVER_CUSTOM_AUTH_ENABLED)) {
        avaticaServerConfiguration = enableCustomAuth(builder, getConf(), ugi);
      } else {
        if (isKerberos) {
          // Enable client auth when using Kerberos auth for HBase
          configureClientAuthentication(builder, disableSpnego, ugi);
        }
        setRemoteUserExtractorIfNecessary(builder, getConf());
        //Avatica doesn't support TLS with custom auth (Why?), hence we only set it in this branch
        setTlsIfNeccessary(builder, getConf());
        setHandler(args, builder);
      }

      enableServerCustomizersIfNecessary(builder, getConf(), avaticaServerConfiguration);

      // Build and start the HttpServer
      server = builder.build();
      server.start();
      if (loadBalancerEnabled) {
        registerToServiceProvider(hostname);
      }
      runningLatch.countDown();
      server.join();
      return 0;
    } catch (Throwable t) {
      LOG.error("Unrecoverable service error. Shutting down.", t);
      this.t = t;
      return -1;
    } finally {
      if (loadBalancerEnabled) {
        unRegister();
      }
    }
  }

  private void setTlsIfNeccessary(Builder<Server> builder, Configuration conf) throws Exception {
    final boolean useTls = getConf().getBoolean(QueryServerProperties.QUERY_SERVER_TLS_ENABLED, QueryServerOptions.DEFAULT_QUERY_SERVER_TLS_ENABLED);
    if(useTls) {
      final String tlsKeystore = getConf().get(QueryServerProperties.QUERY_SERVER_TLS_KEYSTORE);
      final String keystoreType = getConf().get(QueryServerProperties.QUERY_SERVER_TLS_KEYSTORE_TYPE_KEY, QueryServerProperties.QUERY_SERVER_TLS_KEYSTORE_TYPE_DEFAULT);

      final String tlsKeystorePassword = getConf().get(QueryServerProperties.QUERY_SERVER_TLS_KEYSTORE_PASSWORD, QueryServerOptions.DEFAULT_QUERY_SERVER_TLS_KEYSTORE_PASSWORD);
      final String tlsTruststore = getConf().get(QueryServerProperties.QUERY_SERVER_TLS_TRUSTSTORE);
      final String tlsTruststorePassword = getConf().get(QueryServerProperties.QUERY_SERVER_TLS_TRUSTSTORE_PASSWORD, QueryServerOptions.DEFAULT_QUERY_SERVER_TLS_TRUSTSTORE_PASSWORD);
      if(tlsKeystore == null) {
        throw new Exception(String.format("if %s is enabled, %s must be specfified" , QueryServerProperties.QUERY_SERVER_TLS_ENABLED, QueryServerProperties.QUERY_SERVER_TLS_KEYSTORE));
      }
      final File tlsKeystoreFile = new File(tlsKeystore);
      if(tlsTruststore == null) {
        throw new Exception(String.format("if %s is enabled, %s must be specfified" , QueryServerProperties.QUERY_SERVER_TLS_ENABLED, QueryServerProperties.QUERY_SERVER_TLS_TRUSTSTORE));
      }
      final File tlsTruststoreFile = new File(tlsTruststore);
      builder.withTLS(tlsKeystoreFile, tlsKeystorePassword, tlsTruststoreFile, tlsTruststorePassword, keystoreType);
    }
}

  //@VisibleForTesting
  void configureClientAuthentication(final HttpServer.Builder builder, boolean disableSpnego, UserGroupInformation ugi) throws IOException {

    // Enable SPNEGO for client authentication unless it's explicitly disabled
    if (!disableSpnego) {
      configureSpnegoAuthentication(builder, ugi);
    }
    configureCallBack(builder, ugi);
  }

  //@VisibleForTesting
  void configureSpnegoAuthentication(HttpServer.Builder builder, UserGroupInformation ugi) throws IOException {
    String keytabPath = getConf().get(QueryServerProperties.QUERY_SERVER_KEYTAB_FILENAME_ATTRIB);
    File keytab = new File(keytabPath);
    String httpKeytabPath =
            getConf().get(QueryServerProperties.QUERY_SERVER_HTTP_KEYTAB_FILENAME_ATTRIB, null);
    String httpPrincipal = getSpnegoPrincipal(getConf());
    File httpKeytab = null;
    if (null != httpKeytabPath) {
        httpKeytab = new File(httpKeytabPath);
    }

    String realmsString = getConf().get(QueryServerProperties.QUERY_SERVER_KERBEROS_ALLOWED_REALMS, null);
    String[] additionalAllowedRealms = null;
    if (null != realmsString) {
      additionalAllowedRealms = StringUtils.split(realmsString, ',');
    }
    if (null != httpKeytabPath && null != httpPrincipal) {
      builder.withSpnego(httpPrincipal, additionalAllowedRealms).withAutomaticLogin(httpKeytab);
    } else {
      builder.withSpnego(ugi.getUserName(), additionalAllowedRealms)
              .withAutomaticLogin(keytab);
    }
  }

  /**
   * Returns the Kerberos principal to use for SPNEGO, substituting {@code _HOST}
   * if it is present as the "instance" component of the Kerberos principal. It returns
   * the configured principal as-is if {@code _HOST} is not the "instance".
   */
  String getSpnegoPrincipal(Configuration conf) throws IOException {
    String httpPrincipal = conf.get(
        QueryServerProperties.QUERY_SERVER_KERBEROS_HTTP_PRINCIPAL_ATTRIB, null);
    // Backwards compat for a configuration key change
    if (httpPrincipal == null) {
      httpPrincipal = conf.get(
          QueryServerProperties.QUERY_SERVER_KERBEROS_HTTP_PRINCIPAL_ATTRIB_LEGACY, null);
    }

    String hostname = Strings.domainNamePointerToHostName(DNS.getDefaultHost(
        conf.get(QueryServerProperties.QUERY_SERVER_DNS_INTERFACE_ATTRIB, "default"),
        conf.get(QueryServerProperties.QUERY_SERVER_DNS_NAMESERVER_ATTRIB, "default")));
    return SecurityUtil.getServerPrincipal(httpPrincipal, hostname);
  }

  //@VisibleForTesting
  UserGroupInformation getUserGroupInformation() throws IOException {
    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
    LOG.debug("Current user is " + ugi);
    if (!ugi.hasKerberosCredentials()) {
      ugi = UserGroupInformation.getLoginUser();
      LOG.debug("Current user does not have Kerberos credentials, using instead " + ugi);
    }
    return ugi;
  }

  //@VisibleForTesting
  void configureCallBack(HttpServer.Builder<Server> builder, UserGroupInformation ugi) {
    builder.withImpersonation(new PhoenixDoAsCallback(ugi, getConf()));
  }

  private void setHandler(String[] args, HttpServer.Builder<Server> builder) throws Exception {
    Class<? extends PhoenixMetaFactory> factoryClass = getConf().getClass(
            QueryServerProperties.QUERY_SERVER_META_FACTORY_ATTRIB, PhoenixMetaFactoryImpl.class,
            PhoenixMetaFactory.class);
    PhoenixMetaFactory factory =
            factoryClass.getDeclaredConstructor(Configuration.class).newInstance(getConf());
    Meta meta = factory.create(Arrays.asList(args));
    Service service = new LocalService(meta);
    builder.withHandler(service, getSerialization(getConf()));
  }

  public synchronized void stop() {
    server.stop();
  }

  public boolean registerToServiceProvider(String hostName)  {

    boolean success = true ;
    try {
      LoadBalanceZookeeperConf loadBalanceConfiguration = getLoadBalanceConfiguration();
      if (loadBalanceConfiguration == null) {
        throw new NullPointerException();
      }
      this.registry = getRegistry();

      if (registry == null) {
        throw new NullPointerException();
      }
      String zkConnectString = loadBalanceConfiguration.getZkConnectString();
      this.registry.registerServer(loadBalanceConfiguration, getPort(), zkConnectString, hostName);
    } catch(Throwable ex){
      LOG.debug("Caught an error trying to register with the load balancer", ex);
      success = false;
    }
    return success;
  }


  public LoadBalanceZookeeperConf getLoadBalanceConfiguration()  {
    ServiceLoader<LoadBalanceZookeeperConf> serviceLocator= ServiceLoader.load(LoadBalanceZookeeperConf.class);
    LoadBalanceZookeeperConf zookeeperConfig = null;
    try {
      if (serviceLocator.iterator().hasNext())
        zookeeperConfig = serviceLocator.iterator().next();
    } catch(ServiceConfigurationError ex) {
      LOG.debug("Unable to locate the service provider for load balancer configuration", ex);
    }
    return zookeeperConfig;
  }

  public Registry getRegistry()  {
    ServiceLoader<Registry> serviceLocator= ServiceLoader.load(Registry.class);
    Registry registry = null;
    try {
      if (serviceLocator.iterator().hasNext())
        registry = serviceLocator.iterator().next();
    } catch(ServiceConfigurationError ex) {
      LOG.debug("Unable to locate the zookeeper registry for the load balancer", ex);
    }
    return registry;
  }

  public boolean unRegister()  {
    boolean success = true;
    try {
      registry.unRegisterServer();
    }catch(Throwable ex) {
      LOG.debug("Caught an error while de-registering the query server from the load balancer",ex);
      success = false;
    }
    return success;
  }
  /**
   * Parses the serialization method from the configuration.
   *
   * @param conf The configuration to parse
   * @return The Serialization method
   */
  Driver.Serialization getSerialization(Configuration conf) {
    String serializationName = conf.get(QueryServerProperties.QUERY_SERVER_SERIALIZATION_ATTRIB,
        QueryServerOptions.DEFAULT_QUERY_SERVER_SERIALIZATION);

    Driver.Serialization serialization;
    // Otherwise, use what was provided in the configuration
    try {
      serialization = Driver.Serialization.valueOf(serializationName);
    } catch (Exception e) {
      LOG.error("Unknown message serialization type for " + serializationName);
      throw e;
    }

    return serialization;
  }

  @Override public void run() {
    try {
      retCode = run(argv);
    } catch (Exception e) {
      // already logged
    }
  }

  // add remoteUserExtractor to builder if enabled
  //@VisibleForTesting
  public void setRemoteUserExtractorIfNecessary(HttpServer.Builder builder, Configuration conf) {
    if (conf.getBoolean(QueryServerProperties.QUERY_SERVER_WITH_REMOTEUSEREXTRACTOR_ATTRIB,
            QueryServerOptions.DEFAULT_QUERY_SERVER_WITH_REMOTEUSEREXTRACTOR)) {
      builder.withRemoteUserExtractor(createRemoteUserExtractor(conf));
    }
  }

  //@VisibleForTesting
  public void enableServerCustomizersIfNecessary(HttpServer.Builder<Server> builder,
                                                 Configuration conf, AvaticaServerConfiguration avaticaServerConfiguration) {
    // Always try to enable the "provided" ServerCustomizers. The expectation is that the Factory implementation
    // will have toggles for each provided customizer, rather than a global toggle to enable customizers.
    List<ServerCustomizer<Server>> customizers = createServerCustomizers(conf, avaticaServerConfiguration);
    if (customizers != null && !customizers.isEmpty()) {
      builder.withServerCustomizers(customizers, Server.class);
    }
  }

  //@VisibleForTesting
  public AvaticaServerConfiguration enableCustomAuth(HttpServer.Builder<Server> builder,
                                                     Configuration conf, UserGroupInformation ugi) {
    AvaticaServerConfiguration avaticaServerConfiguration = createAvaticaServerConfig(conf, ugi);
    builder.withCustomAuthentication(avaticaServerConfiguration);
    return avaticaServerConfiguration;
  }

  private static final RemoteUserExtractorFactory DEFAULT_USER_EXTRACTOR =
    new RemoteUserExtractorFactory.RemoteUserExtractorFactoryImpl();

  private static final ServerCustomizersFactory DEFAULT_SERVER_CUSTOMIZERS =
    new ServerCustomizersFactory.ServerCustomizersFactoryImpl();

  private static final AvaticaServerConfigurationFactory DEFAULT_SERVER_CONFIG =
    new AvaticaServerConfigurationFactory.AvaticaServerConfigurationFactoryImpl();

  //@VisibleForTesting
  RemoteUserExtractor createRemoteUserExtractor(Configuration conf) {
    RemoteUserExtractorFactory factory =
        InstanceResolver.getSingleton(RemoteUserExtractorFactory.class, DEFAULT_USER_EXTRACTOR);
    return factory.createRemoteUserExtractor(conf);
  }

  //@VisibleForTesting
  List<ServerCustomizer<Server>> createServerCustomizers(Configuration conf, AvaticaServerConfiguration avaticaServerConfiguration) {
    ServerCustomizersFactory factory =
      InstanceResolver.getSingleton(ServerCustomizersFactory.class, DEFAULT_SERVER_CUSTOMIZERS);
    return factory.createServerCustomizers(conf, avaticaServerConfiguration);
  }

  //@VisibleForTesting
  AvaticaServerConfiguration createAvaticaServerConfig(Configuration conf, UserGroupInformation ugi) {
    AvaticaServerConfigurationFactory factory =
            InstanceResolver.getSingleton(AvaticaServerConfigurationFactory.class, DEFAULT_SERVER_CONFIG);
    return factory.getAvaticaServerConfiguration(conf, ugi);
  }

  /**
   * Use the correctly way to extract end user.
   */
  static class PhoenixRemoteUserExtractor implements RemoteUserExtractor{
    private final HttpQueryStringParameterRemoteUserExtractor paramRemoteUserExtractor;
    private final HttpRequestRemoteUserExtractor requestRemoteUserExtractor;
    private final String userExtractParam;

    public PhoenixRemoteUserExtractor(Configuration conf) {
      this.requestRemoteUserExtractor = new HttpRequestRemoteUserExtractor();
      this.userExtractParam = conf.get(QueryServerProperties.QUERY_SERVER_REMOTEUSEREXTRACTOR_PARAM,
              QueryServerOptions.DEFAULT_QUERY_SERVER_REMOTEUSEREXTRACTOR_PARAM);
      this.paramRemoteUserExtractor = new HttpQueryStringParameterRemoteUserExtractor(userExtractParam);
    }

    @Override
    public String extract(HttpServletRequest request) throws RemoteUserExtractionException {
      if (request.getParameter(userExtractParam) != null) {
        String extractedUser = paramRemoteUserExtractor.extract(request);
        UserGroupInformation ugi =
                UserGroupInformation
                        .createRemoteUser(stripHostNameFromPrincipal(request.getRemoteUser()));
        UserGroupInformation proxyUser = UserGroupInformation.createProxyUser(extractedUser, ugi);

        // Check if this user is allowed to be impersonated.
        // Will throw AuthorizationException if the impersonation as this user is not allowed
        try {
          ProxyUsers.authorize(proxyUser, request.getRemoteAddr());
          return extractedUser;
        } catch (AuthorizationException e) {
          throw new RemoteUserExtractionException(e.getMessage(), e);
        }
      } else {
        if (LOG.isDebugEnabled()) {
          LOG.debug("The parameter (" + userExtractParam + ") used to extract the remote user doesn't exist in the request.");
        }
        return requestRemoteUserExtractor.extract(request);
      }

    }
  }

  /**
   * Callback to run the Avatica server action as the remote (proxy) user instead of the server.
   */
  public static class PhoenixDoAsCallback implements DoAsRemoteUserCallback {
    private final UserGroupInformation serverUgi;
    private final SimpleLRUCache<String,UserGroupInformation> ugiCache;

    public PhoenixDoAsCallback(UserGroupInformation serverUgi, Configuration conf) {
      this.serverUgi = Objects.requireNonNull(serverUgi);
      this.ugiCache = new SimpleLRUCache<String,UserGroupInformation>(
              conf.getLong(QueryServerProperties.QUERY_SERVER_UGI_CACHE_MAX_SIZE,
                  QueryServerOptions.DEFAULT_QUERY_SERVER_UGI_CACHE_MAX_SIZE),
              conf.getInt(QueryServerProperties.QUERY_SERVER_UGI_CACHE_CONCURRENCY,
                  QueryServerOptions.DEFAULT_QUERY_SERVER_UGI_CACHE_CONCURRENCY));
    }

    @Override
    public <T> T doAsRemoteUser(String remoteUserName, String remoteAddress,
        final Callable<T> action) throws Exception {
      // We are guaranteed by Avatica that the `remoteUserName` is properly authenticated by the
      // time this method is called. We don't have to verify the wire credentials, we can assume the
      // user provided valid credentials for who it claimed it was.

      // Proxy this user on top of the server's user (the real user). Get a cached instance, the
      // LoadingCache will create a new instance for us if one isn't cached.
      UserGroupInformation proxyUser = createProxyUser(stripHostNameFromPrincipal(remoteUserName));

      // Execute the actual call as this proxy user
      return proxyUser.doAs(new PrivilegedExceptionAction<T>() {
        @Override
        public T run() throws Exception {
          return action.call();
        }
      });
    }

      //@VisibleForTesting
      UserGroupInformation createProxyUser(String remoteUserName) throws ExecutionException {
          // PHOENIX-3164 UGI's hashCode and equals methods rely on reference checks, not
          // value-based checks. We need to make sure we return the same UGI instance for a remote
          // user, otherwise downstream code in Phoenix and HBase may not treat two of the same
          // calls from one user as equivalent.
          return ugiCache.computeIfAbsent(remoteUserName, f -> UserGroupInformation.createProxyUser(f, serverUgi));
      }

      //@VisibleForTesting
      SimpleLRUCache<String,UserGroupInformation> getCache() {
          return ugiCache;
      }
  }

  /**
   * The new Jetty kerberos implementation that we use strips the realm from the principal, which
   * and Hadoop cannot process that.
   * This strips the hostname part as well, so that only the username remains.

   * @param remoteUserName the principal as received from Jetty
   * @return the principal without the hostname part
   */
  //TODO We are probably compensating for a Jetty a bug, which should really be fixed in Jetty
  // See PHOENIX-6913
  private static String stripHostNameFromPrincipal(String remoteUserName) {
      // realm got removed from remoteUserName in CALCITE-4152
      // so we remove the instance name to avoid geting KerberosName$NoMatchingRule exception
      int atSignIndex = remoteUserName.indexOf('@');
      int separatorIndex = remoteUserName.indexOf('/');
      if (atSignIndex == -1 && separatorIndex > 0) {
        remoteUserName = remoteUserName.substring(0, separatorIndex);
      }
      return remoteUserName;
  }

  public static void main(String[] argv) throws Exception {
    int ret = ToolRunner.run(HBaseConfiguration.create(), new QueryServer(), argv);
    System.exit(ret);
  }
}