Repository: hbase Updated Branches: refs/heads/master f0b50a8f9 -> e4b6b4afb
http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/ThriftServer.java ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/ThriftServer.java b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/ThriftServer.java index 5681569..fa3d39d 100644 --- a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/ThriftServer.java +++ b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/ThriftServer.java @@ -18,355 +18,86 @@ */ package org.apache.hadoop.hbase.thrift2; -import java.io.IOException; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.net.UnknownHostException; -import java.security.PrivilegedAction; -import java.util.Map; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; +import static org.apache.hadoop.hbase.thrift.Constants.READONLY_OPTION; +import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_READONLY_ENABLED; +import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_READONLY_ENABLED_DEFAULT; -import javax.security.auth.callback.Callback; -import javax.security.auth.callback.UnsupportedCallbackException; -import javax.security.sasl.AuthorizeCallback; -import javax.security.sasl.SaslServer; +import java.io.IOException; 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.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.CallQueue; -import org.apache.hadoop.hbase.thrift.THBaseThreadPoolExecutor; -import org.apache.hadoop.hbase.thrift.ThriftMetrics; +import org.apache.hadoop.hbase.thrift.HBaseServiceHandler; +import org.apache.hadoop.hbase.thrift.HbaseHandlerMetricsProxy; import org.apache.hadoop.hbase.thrift2.generated.THBaseService; -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.security.SaslRpcServer.SaslGssCallbackHandler; -import org.apache.hadoop.security.UserGroupInformation; -import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.Shell; import org.apache.hadoop.util.ToolRunner; -import org.apache.thrift.TException; import org.apache.thrift.TProcessor; -import org.apache.thrift.protocol.TBinaryProtocol; -import org.apache.thrift.protocol.TCompactProtocol; -import org.apache.thrift.protocol.TProtocol; -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.TThreadPoolServer; -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.TTransportException; -import org.apache.thrift.transport.TTransportFactory; import org.apache.yetus.audience.InterfaceAudience; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -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.Option; -import org.apache.hbase.thirdparty.org.apache.commons.cli.OptionGroup; import org.apache.hbase.thirdparty.org.apache.commons.cli.Options; -import org.apache.hbase.thirdparty.org.apache.commons.cli.ParseException; /** * ThriftServer - this class starts up a Thrift server which implements the HBase API specified in * the HbaseClient.thrift IDL file. */ [email protected](value = "NM_SAME_SIMPLE_NAME_AS_SUPERCLASS", + justification = "Change the name will be an incompatible change, will do it later") @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS) @SuppressWarnings({ "rawtypes", "unchecked" }) -public class ThriftServer extends Configured implements Tool { +public class ThriftServer extends org.apache.hadoop.hbase.thrift.ThriftServer { private static final Logger log = LoggerFactory.getLogger(ThriftServer.class); - /** - * Thrift quality of protection configuration key. Valid values can be: - * privacy: authentication, integrity and confidentiality checking - * integrity: authentication and integrity checking - * authentication: authentication only - * - * This is used to authenticate the callers and support impersonation. - * The thrift server and the HBase cluster must run in secure mode. - */ - static final String THRIFT_QOP_KEY = "hbase.thrift.security.qop"; - - static final String BACKLOG_CONF_KEY = "hbase.regionserver.thrift.backlog"; - public static final int DEFAULT_LISTEN_PORT = 9090; - - private static final String READ_TIMEOUT_OPTION = "readTimeout"; - - /** - * Amount of time in milliseconds before a server thread will timeout - * waiting for client to send data on a connected socket. Currently, - * applies only to TBoundedThreadPoolServer - */ - public static final String THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY = - "hbase.thrift.server.socket.read.timeout"; - public static final int THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT = 60000; - - public ThriftServer() { + public ThriftServer(Configuration conf) { + super(conf); } - private static void printUsage() { + @Override + protected void printUsageAndExit(Options options, int exitCode) + throws Shell.ExitCodeException { HelpFormatter formatter = new HelpFormatter(); - formatter.printHelp("Thrift", null, getOptions(), + formatter.printHelp("Thrift", null, options, "To start the Thrift server run 'hbase-daemon.sh start thrift2' or " + - "'hbase thrift2'\n" + - "To shutdown the thrift server run 'hbase-daemon.sh stop thrift2' or" + - " send a kill signal to the thrift server pid", + "'hbase thrift2'\n" + + "To shutdown the thrift server run 'hbase-daemon.sh stop thrift2' or" + + " send a kill signal to the thrift server pid", true); + throw new Shell.ExitCodeException(exitCode, ""); } - private static Options getOptions() { - Options options = new Options(); - options.addOption("b", "bind", true, - "Address to bind the Thrift server to. [default: 0.0.0.0]"); - options.addOption("p", "port", true, "Port to bind to [default: " + DEFAULT_LISTEN_PORT + "]"); - options.addOption("f", "framed", false, "Use framed transport"); - options.addOption("c", "compact", false, "Use the compact protocol"); - options.addOption("w", "workers", true, "How many worker threads to use."); - options.addOption("s", "selectors", true, "How many selector threads to use."); - options.addOption("q", "callQueueSize", true, - "Max size of request queue (unbounded by default)"); - options.addOption("h", "help", false, "Print help information"); - options.addOption(null, "infoport", true, "Port for web UI"); - 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.addOption("ro", "readonly", false, - "Respond only to read method requests [default: false]"); - OptionGroup servers = new OptionGroup(); - servers.addOption(new Option("nonblocking", false, - "Use the TNonblockingServer. This implies the framed transport.")); - servers.addOption(new Option("hsha", false, - "Use the THsHaServer. This implies the framed transport.")); - servers.addOption(new Option("selector", false, - "Use the TThreadedSelectorServer. This implies the framed transport.")); - servers.addOption(new Option("threadpool", false, - "Use the TThreadPoolServer. This is the default.")); - options.addOptionGroup(servers); - return options; - } - - private static CommandLine parseArguments(Configuration conf, Options options, String[] args) - throws ParseException, IOException { - CommandLineParser parser = new DefaultParser(); - return parser.parse(options, args); - } - - private static TProtocolFactory getTProtocolFactory(boolean isCompact) { - if (isCompact) { - log.debug("Using compact protocol"); - return new TCompactProtocol.Factory(); - } else { - log.debug("Using binary protocol"); - return new TBinaryProtocol.Factory(); - } - } - - private static TTransportFactory getTTransportFactory( - SaslUtil.QualityOfProtection qop, String name, String host, - boolean framed, int frameSize) { - if (framed) { - if (qop != null) { - throw new RuntimeException("Thrift server authentication" - + " doesn't work with framed transport yet"); - } - log.debug("Using framed transport"); - return new TFramedTransport.Factory(frameSize); - } else if (qop == null) { - return new TTransportFactory(); - } else { - Map<String, String> saslProperties = SaslUtil.initSaslProperties(qop.name()); - TSaslServerTransport.Factory saslFactory = new TSaslServerTransport.Factory(); - saslFactory.addServerDefinition("GSSAPI", name, host, saslProperties, - new 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); - } - } - } - }); - return saslFactory; - } - } - - /* - * If bindValue is null, we don't bind. - */ - private static InetSocketAddress bindToPort(String bindValue, int listenPort) - throws UnknownHostException { - try { - if (bindValue == null) { - return new InetSocketAddress(listenPort); - } else { - return new InetSocketAddress(InetAddress.getByName(bindValue), listenPort); - } - } catch (UnknownHostException e) { - throw new RuntimeException("Could not bind to provided ip address", e); - } - } - - private static TServer getTNonBlockingServer(TProtocolFactory protocolFactory, - TProcessor processor, TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) - throws TTransportException { - TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(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); - } - - private static TServer getTHsHaServer(TProtocolFactory protocolFactory, - TProcessor processor, TTransportFactory transportFactory, - int workerThreads, int maxCallQueueSize, - InetSocketAddress inetSocketAddress, ThriftMetrics metrics) - throws TTransportException { - TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress); - log.info("starting HBase HsHA Thrift server on " + inetSocketAddress.toString()); - THsHaServer.Args serverArgs = new THsHaServer.Args(serverTransport); - if (workerThreads > 0) { - // Could support the min & max threads, avoiding to preserve existing functionality. - serverArgs.minWorkerThreads(workerThreads).maxWorkerThreads(workerThreads); - } - ExecutorService executorService = createExecutor( - workerThreads, maxCallQueueSize, metrics); - serverArgs.executorService(executorService); - serverArgs.processor(processor); - serverArgs.transportFactory(transportFactory); - serverArgs.protocolFactory(protocolFactory); - return new THsHaServer(serverArgs); - } - - private static TServer getTThreadedSelectorServer(TProtocolFactory protocolFactory, - TProcessor processor, TTransportFactory transportFactory, - int workerThreads, int selectorThreads, int maxCallQueueSize, - InetSocketAddress inetSocketAddress, ThriftMetrics metrics) - throws TTransportException { - TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress); - log.info("starting HBase ThreadedSelector Thrift server on " + inetSocketAddress.toString()); - TThreadedSelectorServer.Args serverArgs = new TThreadedSelectorServer.Args(serverTransport); - if (workerThreads > 0) { - serverArgs.workerThreads(workerThreads); - } - if (selectorThreads > 0) { - serverArgs.selectorThreads(selectorThreads); - } - - ExecutorService executorService = createExecutor( - workerThreads, maxCallQueueSize, metrics); - serverArgs.executorService(executorService); - serverArgs.processor(processor); - serverArgs.transportFactory(transportFactory); - serverArgs.protocolFactory(protocolFactory); - return new TThreadedSelectorServer(serverArgs); + @Override + protected HBaseServiceHandler createHandler(Configuration conf, UserProvider userProvider) + throws IOException { + return new ThriftHBaseServiceHandler(conf, userProvider); } - private static ExecutorService createExecutor( - int workerThreads, int maxCallQueueSize, ThriftMetrics metrics) { - CallQueue callQueue; - if (maxCallQueueSize > 0) { - callQueue = new CallQueue(new LinkedBlockingQueue<>(maxCallQueueSize), metrics); - } else { - callQueue = new CallQueue(new LinkedBlockingQueue<>(), metrics); - } - - ThreadFactoryBuilder tfb = new ThreadFactoryBuilder(); - tfb.setDaemon(true); - tfb.setNameFormat("thrift2-worker-%d"); - ThreadPoolExecutor pool = new THBaseThreadPoolExecutor(workerThreads, workerThreads, - Long.MAX_VALUE, TimeUnit.SECONDS, callQueue, tfb.build(), metrics); - pool.prestartAllCoreThreads(); - return pool; + @Override + protected TProcessor createProcessor() { + return new THBaseService.Processor<>(HbaseHandlerMetricsProxy + .newInstance((THBaseService.Iface) hbaseServiceHandler, metrics, conf)); } - private static TServer getTThreadPoolServer(TProtocolFactory protocolFactory, - TProcessor processor, - TTransportFactory transportFactory, - int workerThreads, - InetSocketAddress inetSocketAddress, - int backlog, - int clientTimeout, - ThriftMetrics metrics) - throws TTransportException { - TServerTransport serverTransport = new TServerSocket( - new TServerSocket.ServerSocketTransportArgs(). - bindAddr(inetSocketAddress).backlog(backlog). - clientTimeout(clientTimeout)); - log.info("starting HBase ThreadPool Thrift server on " + inetSocketAddress.toString()); - TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport); - serverArgs.processor(processor); - serverArgs.transportFactory(transportFactory); - serverArgs.protocolFactory(protocolFactory); - if (workerThreads > 0) { - serverArgs.maxWorkerThreads(workerThreads); - } - ThreadPoolExecutor executor = new THBaseThreadPoolExecutor(serverArgs.minWorkerThreads, - serverArgs.maxWorkerThreads, serverArgs.stopTimeoutVal, TimeUnit.SECONDS, - new SynchronousQueue<>(), metrics); - serverArgs.executorService(executor); - - return new TThreadPoolServer(serverArgs); + @Override + protected void addOptions(Options options) { + super.addOptions(options); + options.addOption("ro", READONLY_OPTION, false, + "Respond only to read method requests [default: false]"); } - /** - * Adds the option to pre-load filters at startup. - * - * @param conf The current configuration instance. - */ - protected static void registerFilters(Configuration conf) { - String[] filters = conf.getStrings("hbase.thrift.filters"); - if(filters != null) { - for(String filterClass: filters) { - String[] filterPart = filterClass.split(":"); - if(filterPart.length != 2) { - log.warn("Invalid filter specification " + filterClass + " - skipping"); - } else { - ParseFilter.registerFilter(filterPart[0], filterPart[1]); - } - } + @Override + protected void parseCommandLine(CommandLine cmd, Options options) throws Shell.ExitCodeException { + super.parseCommandLine(cmd, options); + boolean readOnly = THRIFT_READONLY_ENABLED_DEFAULT; + if (cmd.hasOption(READONLY_OPTION)) { + readOnly = true; } + conf.setBoolean(THRIFT_READONLY_ENABLED, readOnly); } /** @@ -375,249 +106,8 @@ public class ThriftServer extends Configured implements Tool { public static void main(String[] args) throws Exception { final Configuration conf = HBaseConfiguration.create(); // for now, only time we return is on an argument error. - final int status = ToolRunner.run(conf, new ThriftServer(), args); + final int status = ToolRunner.run(conf, new ThriftServer(conf), args); System.exit(status); } - @Override - public int run(String[] args) throws Exception { - final Configuration conf = getConf(); - Options options = getOptions(); - CommandLine cmd = parseArguments(conf, options, args); - int workerThreads = 0; - int selectorThreads = 0; - int maxCallQueueSize = -1; // use unbounded queue by default - - if (cmd.hasOption("help")) { - printUsage(); - return 1; - } - - // Get address to bind - String bindAddress = getBindAddress(conf, cmd); - - // check if server should only process read requests, if so override the conf - if (cmd.hasOption("readonly")) { - conf.setBoolean("hbase.thrift.readonly", true); - if (log.isDebugEnabled()) { - log.debug("readonly set to true"); - } - } - - // Get read timeout - int readTimeout = getReadTimeout(conf, cmd); - // Get port to bind to - int listenPort = getListenPort(conf, cmd); - // Thrift's implementation uses '0' as a placeholder for 'use the default.' - int backlog = conf.getInt(BACKLOG_CONF_KEY, 0); - - // Local hostname and user name, used only if QOP is configured. - String host = null; - String name = null; - - UserProvider userProvider = UserProvider.instantiate(conf); - // login the server principal (if using secure Hadoop) - boolean securityEnabled = userProvider.isHadoopSecurityEnabled() - && userProvider.isHBaseSecurityEnabled(); - if (securityEnabled) { - host = Strings.domainNamePointerToHostName(DNS.getDefaultHost( - conf.get("hbase.thrift.dns.interface", "default"), - conf.get("hbase.thrift.dns.nameserver", "default"))); - userProvider.login("hbase.thrift.keytab.file", "hbase.thrift.kerberos.principal", host); - } - - UserGroupInformation realUser = userProvider.getCurrent().getUGI(); - String stringQop = conf.get(THRIFT_QOP_KEY); - SaslUtil.QualityOfProtection qop = null; - if (stringQop != null) { - qop = SaslUtil.getQop(stringQop); - if (!securityEnabled) { - throw new IOException("Thrift server must run in secure mode to support authentication"); - } - // Extract the name from the principal - name = SecurityUtil.getUserFromPrincipal(conf.get("hbase.thrift.kerberos.principal")); - } - - boolean nonblocking = cmd.hasOption("nonblocking"); - boolean hsha = cmd.hasOption("hsha"); - boolean selector = cmd.hasOption("selector"); - - ThriftMetrics metrics = new ThriftMetrics(conf, ThriftMetrics.ThriftServerType.TWO); - final JvmPauseMonitor pauseMonitor = new JvmPauseMonitor(conf, metrics.getSource()); - - String implType = getImplType(nonblocking, hsha, selector); - - conf.set("hbase.regionserver.thrift.server.type", implType); - conf.setInt("hbase.regionserver.thrift.port", listenPort); - registerFilters(conf); - - // Construct correct ProtocolFactory - boolean compact = cmd.hasOption("compact") || - conf.getBoolean("hbase.regionserver.thrift.compact", false); - TProtocolFactory protocolFactory = getTProtocolFactory(compact); - final ThriftHBaseServiceHandler hbaseHandler = - new ThriftHBaseServiceHandler(conf, userProvider); - THBaseService.Iface handler = - ThriftHBaseServiceHandler.newInstance(hbaseHandler, metrics); - final THBaseService.Processor p = new THBaseService.Processor(handler); - conf.setBoolean("hbase.regionserver.thrift.compact", compact); - TProcessor processor = p; - - boolean framed = cmd.hasOption("framed") || - conf.getBoolean("hbase.regionserver.thrift.framed", false) || nonblocking || hsha; - TTransportFactory transportFactory = getTTransportFactory(qop, name, host, framed, - conf.getInt("hbase.regionserver.thrift.framed.max_frame_size_in_mb", 2) * 1024 * 1024); - InetSocketAddress inetSocketAddress = bindToPort(bindAddress, listenPort); - conf.setBoolean("hbase.regionserver.thrift.framed", framed); - if (qop != null) { - // Create a processor wrapper, to get the caller - processor = new TProcessor() { - @Override - public boolean process(TProtocol inProt, - TProtocol outProt) throws TException { - TSaslServerTransport saslServerTransport = - (TSaslServerTransport)inProt.getTransport(); - SaslServer saslServer = saslServerTransport.getSaslServer(); - String principal = saslServer.getAuthorizationID(); - hbaseHandler.setEffectiveUser(principal); - return p.process(inProt, outProt); - } - }; - } - - if (cmd.hasOption("w")) { - workerThreads = Integer.parseInt(cmd.getOptionValue("w")); - } - if (cmd.hasOption("s")) { - selectorThreads = Integer.parseInt(cmd.getOptionValue("s")); - } - if (cmd.hasOption("q")) { - maxCallQueueSize = Integer.parseInt(cmd.getOptionValue("q")); - } - - // check for user-defined info server port setting, if so override the conf - try { - if (cmd.hasOption("infoport")) { - String val = cmd.getOptionValue("infoport"); - conf.setInt("hbase.thrift.info.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", e); - printUsage(); - System.exit(1); - } - - // Put up info server. - startInfoServer(conf); - - final TServer tserver = getServer(workerThreads, selectorThreads, maxCallQueueSize, readTimeout, - backlog, nonblocking, hsha, selector, metrics, protocolFactory, processor, - transportFactory, inetSocketAddress); - - realUser.doAs( - new PrivilegedAction<Object>() { - @Override - public Object run() { - pauseMonitor.start(); - try { - tserver.serve(); - return null; - } finally { - pauseMonitor.stop(); - } - } - }); - // when tserver.stop eventually happens we'll get here. - return 0; - } - - private String getImplType(boolean nonblocking, boolean hsha, boolean selector) { - String implType = "threadpool"; - - if (nonblocking) { - implType = "nonblocking"; - } else if (hsha) { - implType = "hsha"; - } else if (selector) { - implType = "selector"; - } - - return implType; - } - - private String getBindAddress(Configuration conf, CommandLine cmd) { - String bindAddress; - if (cmd.hasOption("bind")) { - bindAddress = cmd.getOptionValue("bind"); - conf.set("hbase.thrift.info.bindAddress", bindAddress); - } else { - bindAddress = conf.get("hbase.thrift.info.bindAddress"); - } - return bindAddress; - } - - private int getListenPort(Configuration conf, CommandLine cmd) { - int listenPort; - try { - if (cmd.hasOption("port")) { - listenPort = Integer.parseInt(cmd.getOptionValue("port")); - } else { - listenPort = conf.getInt("hbase.regionserver.thrift.port", DEFAULT_LISTEN_PORT); - } - } catch (NumberFormatException e) { - throw new RuntimeException("Could not parse the value provided for the port option", e); - } - return listenPort; - } - - private int getReadTimeout(Configuration conf, CommandLine cmd) { - int readTimeout; - if (cmd.hasOption(READ_TIMEOUT_OPTION)) { - try { - readTimeout = Integer.parseInt(cmd.getOptionValue(READ_TIMEOUT_OPTION)); - } catch (NumberFormatException e) { - throw new RuntimeException("Could not parse the value provided for the timeout option", e); - } - } else { - readTimeout = conf.getInt(THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY, - THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT); - } - return readTimeout; - } - - private void startInfoServer(Configuration conf) throws IOException { - int port = conf.getInt("hbase.thrift.info.port", 9095); - - if (port >= 0) { - conf.setLong("startcode", System.currentTimeMillis()); - String a = conf.get("hbase.thrift.info.bindAddress", "0.0.0.0"); - InfoServer infoServer = new InfoServer("thrift", a, port, false, conf); - infoServer.setAttribute("hbase.conf", conf); - infoServer.start(); - } - } - - private TServer getServer(int workerThreads, int selectorThreads, int maxCallQueueSize, - int readTimeout, int backlog, boolean nonblocking, boolean hsha, boolean selector, - ThriftMetrics metrics, TProtocolFactory protocolFactory, TProcessor processor, - TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) - throws TTransportException { - TServer server; - - if (nonblocking) { - server = getTNonBlockingServer(protocolFactory, processor, transportFactory, - inetSocketAddress); - } else if (hsha) { - server = getTHsHaServer(protocolFactory, processor, transportFactory, workerThreads, - maxCallQueueSize, inetSocketAddress, metrics); - } else if (selector) { - server = getTThreadedSelectorServer(protocolFactory, processor, transportFactory, - workerThreads, selectorThreads, maxCallQueueSize, inetSocketAddress, metrics); - } else { - server = getTThreadPoolServer(protocolFactory, processor, transportFactory, workerThreads, - inetSocketAddress, backlog, readTimeout, metrics); - } - return server; - } } http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/main/resources/hbase-webapps/thrift/thrift.jsp ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/main/resources/hbase-webapps/thrift/thrift.jsp b/hbase-thrift/src/main/resources/hbase-webapps/thrift/thrift.jsp index eb329fe..b0064b4 100644 --- a/hbase-thrift/src/main/resources/hbase-webapps/thrift/thrift.jsp +++ b/hbase-thrift/src/main/resources/hbase-webapps/thrift/thrift.jsp @@ -20,10 +20,10 @@ <%@ page contentType="text/html;charset=UTF-8" import="org.apache.hadoop.conf.Configuration" import="org.apache.hadoop.hbase.HBaseConfiguration" - import="org.apache.hadoop.hbase.thrift.ThriftServerRunner.ImplType" import="org.apache.hadoop.hbase.util.VersionInfo" import="java.util.Date" %> +<%@ page import="org.apache.hadoop.hbase.thrift.ImplType" %> <% Configuration conf = (Configuration)getServletContext().getAttribute("hbase.conf"); http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftHttpServer.java ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftHttpServer.java b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftHttpServer.java index 0eb7641..fe6655e 100644 --- a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftHttpServer.java +++ b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftHttpServer.java @@ -17,8 +17,9 @@ */ package org.apache.hadoop.hbase.thrift; +import static org.apache.hadoop.hbase.thrift.Constants.INFOPORT_OPTION; +import static org.apache.hadoop.hbase.thrift.Constants.PORT_OPTION; import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNull; import static org.junit.Assert.fail; import java.net.HttpURLConnection; @@ -65,7 +66,7 @@ public class TestThriftHttpServer { private static final Logger LOG = LoggerFactory.getLogger(TestThriftHttpServer.class); - static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); + protected static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); private Thread httpServerThread; private volatile Exception httpServerException; @@ -97,18 +98,17 @@ public class TestThriftHttpServer { conf.set("hbase.thrift.security.qop", "privacy"); conf.setBoolean("hbase.thrift.ssl.enabled", false); - ThriftServerRunner runner = null; + ThriftServer server = null; ExpectedException thrown = ExpectedException.none(); try { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Thrift HTTP Server's QoP is privacy, " + "but hbase.thrift.ssl.enabled is false"); - runner = new ThriftServerRunner(conf); + server = new ThriftServer(conf); + server.run(); fail("Thrift HTTP Server starts up even with wrong security configurations."); } catch (Exception e) { } - - assertNull(runner); } private void startHttpServerThread(final String[] args) { @@ -117,7 +117,7 @@ public class TestThriftHttpServer { httpServerException = null; httpServerThread = new Thread(() -> { try { - thriftServer.doMain(args); + thriftServer.run(args); } catch (Exception e) { httpServerException = e; } @@ -145,6 +145,10 @@ public class TestThriftHttpServer { runThriftServer(1024 * 64); } + protected ThriftServer createThriftServer() { + return new ThriftServer(TEST_UTIL.getConfiguration()); + } + @Test public void testRunThriftServer() throws Exception { runThriftServer(0); @@ -153,14 +157,14 @@ public class TestThriftHttpServer { void runThriftServer(int customHeaderSize) throws Exception { List<String> args = new ArrayList<>(3); port = HBaseTestingUtility.randomFreePort(); - args.add("-" + ThriftServer.PORT_OPTION); + args.add("-" + PORT_OPTION); args.add(String.valueOf(port)); - args.add("-" + ThriftServer.INFOPORT_OPTION); + args.add("-" + INFOPORT_OPTION); int infoPort = HBaseTestingUtility.randomFreePort(); args.add(String.valueOf(infoPort)); args.add("start"); - thriftServer = new ThriftServer(TEST_UTIL.getConfiguration()); + thriftServer = createThriftServer(); startHttpServerThread(args.toArray(new String[args.size()])); // wait up to 10s for the server to start @@ -195,9 +199,9 @@ public class TestThriftHttpServer { Assert.assertEquals(HttpURLConnection.HTTP_FORBIDDEN, conn.getResponseCode()); } - static volatile boolean tableCreated = false; + protected static volatile boolean tableCreated = false; - void talkToThriftServer(String url, int customHeaderSize) throws Exception { + protected void talkToThriftServer(String url, int customHeaderSize) throws Exception { THttpClient httpClient = new THttpClient(url); httpClient.open(); http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServer.java ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServer.java b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServer.java index 05dc2ae..09d0605 100644 --- a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServer.java +++ b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServer.java @@ -17,6 +17,7 @@ */ package org.apache.hadoop.hbase.thrift; +import static org.apache.hadoop.hbase.thrift.Constants.COALESCE_INC_KEY; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -46,7 +47,6 @@ import org.apache.hadoop.hbase.security.UserProvider; import org.apache.hadoop.hbase.test.MetricsAssertHelper; import org.apache.hadoop.hbase.testclassification.ClientTests; import org.apache.hadoop.hbase.testclassification.LargeTests; -import org.apache.hadoop.hbase.thrift.ThriftServerRunner.HBaseHandler; import org.apache.hadoop.hbase.thrift.generated.BatchMutation; import org.apache.hadoop.hbase.thrift.generated.ColumnDescriptor; import org.apache.hadoop.hbase.thrift.generated.Hbase; @@ -71,7 +71,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Unit testing for ThriftServerRunner.HBaseHandler, a part of the + * Unit testing for ThriftServerRunner.HBaseServiceHandler, a part of the * org.apache.hadoop.hbase.thrift package. */ @Category({ClientTests.class, LargeTests.class}) @@ -113,7 +113,7 @@ public class TestThriftServer { @BeforeClass public static void beforeClass() throws Exception { - UTIL.getConfiguration().setBoolean(ThriftServerRunner.COALESCE_INC_KEY, true); + UTIL.getConfiguration().setBoolean(COALESCE_INC_KEY, true); UTIL.getConfiguration().setBoolean("hbase.table.sanity.checks", false); UTIL.getConfiguration().setInt("hbase.client.retries.number", 3); UTIL.startMiniCluster(); @@ -152,8 +152,8 @@ public class TestThriftServer { * IllegalArgument exception. */ public void doTestTableCreateDrop() throws Exception { - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); doTestTableCreateDrop(handler); } @@ -163,7 +163,7 @@ public class TestThriftServer { dropTestTables(handler); } - public static final class MySlowHBaseHandler extends ThriftServerRunner.HBaseHandler + public static final class MySlowHBaseHandler extends ThriftHBaseServiceHandler implements Hbase.Iface { protected MySlowHBaseHandler(Configuration c) @@ -230,7 +230,7 @@ public class TestThriftServer { private static Hbase.Iface getHandlerForMetricsTest(ThriftMetrics metrics, Configuration conf) throws Exception { Hbase.Iface handler = new MySlowHBaseHandler(conf); - return HbaseHandlerMetricsProxy.newInstance(handler, metrics, conf); + return HbaseHandlerMetricsProxy.newInstance((ThriftHBaseServiceHandler)handler, metrics, conf); } private static ThriftMetrics getMetrics(Configuration conf) throws Exception { @@ -270,15 +270,15 @@ public class TestThriftServer { } public void doTestIncrements() throws Exception { - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); createTestTables(handler); doTestIncrements(handler); dropTestTables(handler); } - public static void doTestIncrements(HBaseHandler handler) throws Exception { + public static void doTestIncrements(ThriftHBaseServiceHandler handler) throws Exception { List<Mutation> mutations = new ArrayList<>(1); mutations.add(new Mutation(false, columnAAname, valueEname, true)); mutations.add(new Mutation(false, columnAname, valueEname, true)); @@ -318,8 +318,8 @@ public class TestThriftServer { * versions. */ public void doTestTableMutations() throws Exception { - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); doTestTableMutations(handler); } @@ -395,8 +395,8 @@ public class TestThriftServer { */ public void doTestTableTimestampsAndColumns() throws Exception { // Setup - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); handler.createTable(tableAname, getColumnDescriptors()); @@ -473,8 +473,8 @@ public class TestThriftServer { */ public void doTestTableScanners() throws Exception { // Setup - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); handler.createTable(tableAname, getColumnDescriptors()); @@ -592,8 +592,8 @@ public class TestThriftServer { * Tests for GetTableRegions */ public void doTestGetTableRegions() throws Exception { - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); doTestGetTableRegions(handler); } @@ -620,7 +620,7 @@ public class TestThriftServer { conf.set("hbase.thrift.filters", "MyFilter:filterclass"); - ThriftServerRunner.registerFilters(conf); + ThriftServer.registerFilters(conf); Map<String, String> registeredFilters = ParseFilter.getAllFilters(); @@ -628,8 +628,8 @@ public class TestThriftServer { } public void doTestGetRegionInfo() throws Exception { - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); doTestGetRegionInfo(handler); } @@ -655,8 +655,8 @@ public class TestThriftServer { * Appends the value to a cell and checks that the cell value is updated properly. */ public static void doTestAppend() throws Exception { - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); handler.createTable(tableAname, getColumnDescriptors()); try { @@ -687,8 +687,8 @@ public class TestThriftServer { * the checkAndPut succeeds. */ public static void doTestCheckAndPut() throws Exception { - ThriftServerRunner.HBaseHandler handler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler handler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); handler.createTable(tableAname, getColumnDescriptors()); try { @@ -729,8 +729,8 @@ public class TestThriftServer { Configuration conf = UTIL.getConfiguration(); ThriftMetrics metrics = getMetrics(conf); - ThriftServerRunner.HBaseHandler hbaseHandler = - new ThriftServerRunner.HBaseHandler(UTIL.getConfiguration(), + ThriftHBaseServiceHandler hbaseHandler = + new ThriftHBaseServiceHandler(UTIL.getConfiguration(), UserProvider.instantiate(UTIL.getConfiguration())); Hbase.Iface handler = HbaseHandlerMetricsProxy.newInstance(hbaseHandler, metrics, conf); @@ -863,10 +863,10 @@ public class TestThriftServer { * the scanner. * * @param scannerId the scanner to close - * @param handler the HBaseHandler interfacing to HBase + * @param handler the HBaseServiceHandler interfacing to HBase */ private void closeScanner( - int scannerId, ThriftServerRunner.HBaseHandler handler) throws Exception { + int scannerId, ThriftHBaseServiceHandler handler) throws Exception { handler.scannerGet(scannerId); handler.scannerClose(scannerId); } http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerCmdLine.java ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerCmdLine.java b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerCmdLine.java index 9e5a103..08dae61 100644 --- a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerCmdLine.java +++ b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerCmdLine.java @@ -17,6 +17,11 @@ */ package org.apache.hadoop.hbase.thrift; +import static org.apache.hadoop.hbase.thrift.Constants.BIND_OPTION; +import static org.apache.hadoop.hbase.thrift.Constants.COMPACT_OPTION; +import static org.apache.hadoop.hbase.thrift.Constants.FRAMED_OPTION; +import static org.apache.hadoop.hbase.thrift.Constants.INFOPORT_OPTION; +import static org.apache.hadoop.hbase.thrift.Constants.PORT_OPTION; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -28,7 +33,6 @@ import org.apache.hadoop.hbase.HBaseClassTestRule; import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.testclassification.ClientTests; import org.apache.hadoop.hbase.testclassification.LargeTests; -import org.apache.hadoop.hbase.thrift.ThriftServerRunner.ImplType; import org.apache.hadoop.hbase.thrift.generated.Hbase; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper; @@ -68,12 +72,12 @@ public class TestThriftServerCmdLine { private static final Logger LOG = LoggerFactory.getLogger(TestThriftServerCmdLine.class); - private final ImplType implType; - private boolean specifyFramed; - private boolean specifyBindIP; - private boolean specifyCompact; + protected final ImplType implType; + protected boolean specifyFramed; + protected boolean specifyBindIP; + protected boolean specifyCompact; - private static final HBaseTestingUtility TEST_UTIL = + protected static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); private Thread cmdLineThread; @@ -81,8 +85,8 @@ public class TestThriftServerCmdLine { private Exception clientSideException; - private ThriftServer thriftServer; - private int port; + private volatile ThriftServer thriftServer; + protected int port; @Parameters public static Collection<Object[]> getParameters() { @@ -143,8 +147,9 @@ public class TestThriftServerCmdLine { @Override public void run() { try { - thriftServer.doMain(args); + thriftServer.run(args); } catch (Exception e) { + LOG.error("Error when start thrift server", e); cmdLineException = e; } } @@ -154,6 +159,10 @@ public class TestThriftServerCmdLine { cmdLineThread.start(); } + protected ThriftServer createThriftServer() { + return new ThriftServer(TEST_UTIL.getConfiguration()); + } + @Test public void testRunThriftServer() throws Exception { List<String> args = new ArrayList<>(); @@ -163,37 +172,37 @@ public class TestThriftServerCmdLine { args.add(serverTypeOption); } port = HBaseTestingUtility.randomFreePort(); - args.add("-" + ThriftServer.PORT_OPTION); + args.add("-" + PORT_OPTION); args.add(String.valueOf(port)); - args.add("-" + ThriftServer.INFOPORT_OPTION); + args.add("-" + INFOPORT_OPTION); int infoPort = HBaseTestingUtility.randomFreePort(); args.add(String.valueOf(infoPort)); if (specifyFramed) { - args.add("-" + ThriftServer.FRAMED_OPTION); + args.add("-" + FRAMED_OPTION); } if (specifyBindIP) { - args.add("-" + ThriftServer.BIND_OPTION); + args.add("-" + BIND_OPTION); args.add(InetAddress.getLocalHost().getHostName()); } if (specifyCompact) { - args.add("-" + ThriftServer.COMPACT_OPTION); + args.add("-" + COMPACT_OPTION); } args.add("start"); - thriftServer = new ThriftServer(TEST_UTIL.getConfiguration()); + thriftServer = createThriftServer(); startCmdLineThread(args.toArray(new String[args.size()])); // wait up to 10s for the server to start for (int i = 0; i < 100 - && (thriftServer.serverRunner == null || thriftServer.serverRunner.tserver == null); i++) { + && (thriftServer.tserver == null); i++) { Thread.sleep(100); } Class<? extends TServer> expectedClass = implType != null ? implType.serverClass : TBoundedThreadPoolServer.class; assertEquals(expectedClass, - thriftServer.serverRunner.tserver.getClass()); + thriftServer.tserver.getClass()); try { talkToThriftServer(); @@ -210,9 +219,9 @@ public class TestThriftServerCmdLine { } } - private static volatile boolean tableCreated = false; + protected static volatile boolean tableCreated = false; - private void talkToThriftServer() throws Exception { + protected void talkToThriftServer() throws Exception { TSocket sock = new TSocket(InetAddress.getLocalHost().getHostName(), port); TTransport transport = sock; @@ -228,6 +237,7 @@ public class TestThriftServerCmdLine { } else { prot = new TBinaryProtocol(transport); } + Hbase.Client client = new Hbase.Client(prot); if (!tableCreated){ TestThriftServer.createTestTables(client); http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftSpnegoHttpServer.java ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftSpnegoHttpServer.java b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftSpnegoHttpServer.java index 49eafec..45abe79 100644 --- a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftSpnegoHttpServer.java +++ b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftSpnegoHttpServer.java @@ -18,12 +18,7 @@ */ package org.apache.hadoop.hbase.thrift; -import static org.apache.hadoop.hbase.thrift.ThriftServerRunner.THRIFT_KERBEROS_PRINCIPAL_KEY; -import static org.apache.hadoop.hbase.thrift.ThriftServerRunner.THRIFT_KEYTAB_FILE_KEY; -import static org.apache.hadoop.hbase.thrift.ThriftServerRunner.THRIFT_SPNEGO_KEYTAB_FILE_KEY; -import static org.apache.hadoop.hbase.thrift.ThriftServerRunner.THRIFT_SPNEGO_PRINCIPAL_KEY; -import static org.apache.hadoop.hbase.thrift.ThriftServerRunner.THRIFT_SUPPORT_PROXYUSER_KEY; -import static org.apache.hadoop.hbase.thrift.ThriftServerRunner.USE_HTTP_CONF_KEY; +import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SUPPORT_PROXYUSER_KEY; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; @@ -133,14 +128,14 @@ public class TestThriftSpnegoHttpServer extends TestThriftHttpServer { HBaseKerberosUtils.setSecuredConfiguration(conf, serverPrincipal, spnegoServerPrincipal); conf.setBoolean(THRIFT_SUPPORT_PROXYUSER_KEY, true); - conf.setBoolean(USE_HTTP_CONF_KEY, true); + conf.setBoolean(Constants.USE_HTTP_CONF_KEY, true); conf.set("hadoop.proxyuser.hbase.hosts", "*"); conf.set("hadoop.proxyuser.hbase.groups", "*"); - conf.set(THRIFT_KERBEROS_PRINCIPAL_KEY, serverPrincipal); - conf.set(THRIFT_KEYTAB_FILE_KEY, serverKeytab.getAbsolutePath()); - conf.set(THRIFT_SPNEGO_PRINCIPAL_KEY, spnegoServerPrincipal); - conf.set(THRIFT_SPNEGO_KEYTAB_FILE_KEY, spnegoServerKeytab.getAbsolutePath()); + conf.set(Constants.THRIFT_KERBEROS_PRINCIPAL_KEY, serverPrincipal); + conf.set(Constants.THRIFT_KEYTAB_FILE_KEY, serverKeytab.getAbsolutePath()); + conf.set(Constants.THRIFT_SPNEGO_PRINCIPAL_KEY, spnegoServerPrincipal); + conf.set(Constants.THRIFT_SPNEGO_KEYTAB_FILE_KEY, spnegoServerKeytab.getAbsolutePath()); } @BeforeClass @@ -170,7 +165,7 @@ public class TestThriftSpnegoHttpServer extends TestThriftHttpServer { spnegoServerKeytab = new File(keytabDir, spnegoServerPrincipal.replace('/', '_') + ".keytab"); setupUser(kdc, spnegoServerKeytab, spnegoServerPrincipal); - TEST_UTIL.getConfiguration().setBoolean(USE_HTTP_CONF_KEY, true); + TEST_UTIL.getConfiguration().setBoolean(Constants.USE_HTTP_CONF_KEY, true); TEST_UTIL.getConfiguration().setBoolean("hbase.table.sanity.checks", false); addSecurityConfigurations(TEST_UTIL.getConfiguration()); @@ -191,7 +186,7 @@ public class TestThriftSpnegoHttpServer extends TestThriftHttpServer { } @Override - void talkToThriftServer(String url, int customHeaderSize) throws Exception { + protected void talkToThriftServer(String url, int customHeaderSize) throws Exception { // Close httpClient and THttpClient automatically on any failures try ( CloseableHttpClient httpClient = createHttpClient(); http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThrift2HttpServer.java ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThrift2HttpServer.java b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThrift2HttpServer.java new file mode 100644 index 0000000..cf084c9 --- /dev/null +++ b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThrift2HttpServer.java @@ -0,0 +1,90 @@ +/** + * 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.thrift2; + +import java.util.ArrayList; + +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.testclassification.ClientTests; +import org.apache.hadoop.hbase.testclassification.MediumTests; +import org.apache.hadoop.hbase.thrift.TestThriftHttpServer; +import org.apache.hadoop.hbase.thrift2.generated.TColumnFamilyDescriptor; +import org.apache.hadoop.hbase.thrift2.generated.THBaseService; +import org.apache.hadoop.hbase.thrift2.generated.TTableDescriptor; +import org.apache.hadoop.hbase.thrift2.generated.TTableName; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.transport.THttpClient; +import org.junit.Assert; +import org.junit.ClassRule; +import org.junit.experimental.categories.Category; + +@Category({ ClientTests.class, MediumTests.class}) +public class TestThrift2HttpServer extends TestThriftHttpServer { + private static final String TABLENAME = "TestThrift2HttpServerTable"; + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestThrift2HttpServer.class); + + + + @Override + protected ThriftServer createThriftServer() { + return new ThriftServer(TEST_UTIL.getConfiguration()); + } + + @Override + protected void talkToThriftServer(String url, int customHeaderSize) throws Exception { + THttpClient httpClient = new THttpClient(url); + httpClient.open(); + + if (customHeaderSize > 0) { + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < customHeaderSize; i++) { + sb.append("a"); + } + httpClient.setCustomHeader("User-Agent", sb.toString()); + } + + try { + TProtocol prot; + prot = new TBinaryProtocol(httpClient); + THBaseService.Client client = new THBaseService.Client(prot); + TTableName tTableName = new TTableName(); + tTableName.setNs(Bytes.toBytes("")); + tTableName.setQualifier(Bytes.toBytes(TABLENAME)); + if (!tableCreated){ + Assert.assertTrue(!client.tableExists(tTableName)); + TTableDescriptor tTableDescriptor = new TTableDescriptor(); + tTableDescriptor.setTableName(tTableName); + TColumnFamilyDescriptor columnFamilyDescriptor = new TColumnFamilyDescriptor(); + columnFamilyDescriptor.setName(Bytes.toBytes(TABLENAME)); + tTableDescriptor.addToColumns(columnFamilyDescriptor); + client.createTable(tTableDescriptor, new ArrayList<>()); + tableCreated = true; + } + Assert.assertTrue(client.tableExists(tTableName)); + } finally { + httpClient.close(); + } + } + + +} http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThrift2ServerCmdLine.java ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThrift2ServerCmdLine.java b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThrift2ServerCmdLine.java new file mode 100644 index 0000000..7489bd7 --- /dev/null +++ b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThrift2ServerCmdLine.java @@ -0,0 +1,99 @@ +/** + * 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.thrift2; + +import java.net.InetAddress; +import java.util.ArrayList; + +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.testclassification.ClientTests; +import org.apache.hadoop.hbase.testclassification.MediumTests; +import org.apache.hadoop.hbase.thrift.ImplType; +import org.apache.hadoop.hbase.thrift.TestThriftServerCmdLine; +import org.apache.hadoop.hbase.thrift2.generated.TColumnFamilyDescriptor; +import org.apache.hadoop.hbase.thrift2.generated.THBaseService; +import org.apache.hadoop.hbase.thrift2.generated.TTableDescriptor; +import org.apache.hadoop.hbase.thrift2.generated.TTableName; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TCompactProtocol; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.transport.TFramedTransport; +import org.apache.thrift.transport.TSocket; +import org.apache.thrift.transport.TTransport; +import org.junit.Assert; +import org.junit.ClassRule; +import org.junit.experimental.categories.Category; + +@Category({ ClientTests.class, MediumTests.class}) +public class TestThrift2ServerCmdLine extends TestThriftServerCmdLine { + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestThrift2ServerCmdLine.class); + + private static final String TABLENAME = "TestThrift2ServerCmdLineTable"; + + + @Override + protected ThriftServer createThriftServer() { + return new ThriftServer(TEST_UTIL.getConfiguration()); + } + + public TestThrift2ServerCmdLine(ImplType implType, boolean specifyFramed, + boolean specifyBindIP, boolean specifyCompact) { + super(implType, specifyFramed, specifyBindIP, specifyCompact); + } + + @Override + protected void talkToThriftServer() throws Exception { + TSocket sock = new TSocket(InetAddress.getLocalHost().getHostName(), + port); + TTransport transport = sock; + if (specifyFramed || implType.isAlwaysFramed()) { + transport = new TFramedTransport(transport); + } + + sock.open(); + try { + TProtocol tProtocol; + if (specifyCompact) { + tProtocol = new TCompactProtocol(transport); + } else { + tProtocol = new TBinaryProtocol(transport); + } + THBaseService.Client client = new THBaseService.Client(tProtocol); + TTableName tTableName = new TTableName(); + tTableName.setNs(Bytes.toBytes("")); + tTableName.setQualifier(Bytes.toBytes(TABLENAME)); + if (!tableCreated){ + Assert.assertTrue(!client.tableExists(tTableName)); + TTableDescriptor tTableDescriptor = new TTableDescriptor(); + tTableDescriptor.setTableName(tTableName); + TColumnFamilyDescriptor columnFamilyDescriptor = new TColumnFamilyDescriptor(); + columnFamilyDescriptor.setName(Bytes.toBytes(TABLENAME)); + tTableDescriptor.addToColumns(columnFamilyDescriptor); + client.createTable(tTableDescriptor, new ArrayList<>()); + tableCreated = true; + } + Assert.assertTrue(client.tableExists(tTableName)); + } finally { + sock.close(); + } + } + +} http://git-wip-us.apache.org/repos/asf/hbase/blob/e4b6b4af/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThriftHBaseServiceHandler.java ---------------------------------------------------------------------- diff --git a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThriftHBaseServiceHandler.java b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThriftHBaseServiceHandler.java index 0734f85..844c1c0 100644 --- a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThriftHBaseServiceHandler.java +++ b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThriftHBaseServiceHandler.java @@ -18,6 +18,8 @@ package org.apache.hadoop.hbase.thrift2; import static java.nio.ByteBuffer.wrap; +import static org.apache.hadoop.hbase.thrift.HBaseServiceHandler.CLEANUP_INTERVAL; +import static org.apache.hadoop.hbase.thrift.HBaseServiceHandler.MAX_IDLETIME; import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.deleteFromThrift; import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.getFromThrift; import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.incrementFromThrift; @@ -71,6 +73,7 @@ import org.apache.hadoop.hbase.test.MetricsAssertHelper; import org.apache.hadoop.hbase.testclassification.ClientTests; import org.apache.hadoop.hbase.testclassification.MediumTests; import org.apache.hadoop.hbase.thrift.ErrorThrowingGetObserver; +import org.apache.hadoop.hbase.thrift.HbaseHandlerMetricsProxy; import org.apache.hadoop.hbase.thrift.ThriftMetrics; import org.apache.hadoop.hbase.thrift2.generated.TAppend; import org.apache.hadoop.hbase.thrift2.generated.TColumn; @@ -115,7 +118,7 @@ import org.apache.hbase.thirdparty.com.google.common.collect.Lists; import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils; /** - * Unit testing for ThriftServer.HBaseHandler, a part of the org.apache.hadoop.hbase.thrift2 + * Unit testing for ThriftServer.HBaseServiceHandler, a part of the org.apache.hadoop.hbase.thrift2 * package. */ @Category({ClientTests.class, MediumTests.class}) @@ -749,8 +752,8 @@ public class TestThriftHBaseServiceHandler { int cleanUpInterval = 100; Configuration conf = new Configuration(UTIL.getConfiguration()); // Set the ConnectionCache timeout to trigger halfway through the trials - conf.setInt(ThriftHBaseServiceHandler.MAX_IDLETIME, (numTrials / 2) * trialPause); - conf.setInt(ThriftHBaseServiceHandler.CLEANUP_INTERVAL, cleanUpInterval); + conf.setInt(MAX_IDLETIME, (numTrials / 2) * trialPause); + conf.setInt(CLEANUP_INTERVAL, cleanUpInterval); ThriftHBaseServiceHandler handler = new ThriftHBaseServiceHandler(conf, UserProvider.instantiate(conf)); @@ -1206,7 +1209,7 @@ public class TestThriftHBaseServiceHandler { ThriftMetrics metrics = getMetrics(conf); ThriftHBaseServiceHandler hbaseHandler = createHandler(); THBaseService.Iface handler = - ThriftHBaseServiceHandler.newInstance(hbaseHandler, metrics); + HbaseHandlerMetricsProxy.newInstance(hbaseHandler, metrics, conf); byte[] rowName = Bytes.toBytes("testMetrics"); ByteBuffer table = wrap(tableAname); @@ -1249,7 +1252,7 @@ public class TestThriftHBaseServiceHandler { ThriftHBaseServiceHandler hbaseHandler = createHandler(); ThriftMetrics metrics = getMetrics(UTIL.getConfiguration()); THBaseService.Iface handler = - ThriftHBaseServiceHandler.newInstance(hbaseHandler, metrics); + HbaseHandlerMetricsProxy.newInstance(hbaseHandler, metrics, null); ByteBuffer tTableName = wrap(tableName.getName()); // check metrics increment with a successful get @@ -1323,7 +1326,7 @@ public class TestThriftHBaseServiceHandler { ThriftHBaseServiceHandler hbaseHandler = createHandler(); ThriftMetrics metrics = getMetrics(UTIL.getConfiguration()); THBaseService.Iface handler = - ThriftHBaseServiceHandler.newInstance(hbaseHandler, metrics); + HbaseHandlerMetricsProxy.newInstance(hbaseHandler, metrics, null); ByteBuffer tTableName = wrap(tableName.getName()); // check metrics latency with a successful get
