1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.thrift;
20
21 import static org.apache.hadoop.hbase.util.Bytes.getBytes;
22
23 import java.io.IOException;
24 import java.net.InetAddress;
25 import java.net.InetSocketAddress;
26 import java.net.UnknownHostException;
27 import java.nio.ByteBuffer;
28 import java.security.PrivilegedAction;
29 import java.util.ArrayList;
30 import java.util.Arrays;
31 import java.util.Collections;
32 import java.util.HashMap;
33 import java.util.List;
34 import java.util.Map;
35 import java.util.TreeMap;
36 import java.util.concurrent.BlockingQueue;
37 import java.util.concurrent.ExecutorService;
38 import java.util.concurrent.LinkedBlockingQueue;
39 import java.util.concurrent.TimeUnit;
40
41 import javax.security.auth.callback.Callback;
42 import javax.security.auth.callback.UnsupportedCallbackException;
43 import javax.security.sasl.AuthorizeCallback;
44 import javax.security.sasl.SaslServer;
45
46 import org.apache.commons.cli.CommandLine;
47 import org.apache.commons.cli.Option;
48 import org.apache.commons.cli.OptionGroup;
49 import org.apache.commons.logging.Log;
50 import org.apache.commons.logging.LogFactory;
51 import org.apache.hadoop.conf.Configuration;
52 import org.apache.hadoop.hbase.HBaseConfiguration;
53 import org.apache.hadoop.hbase.HColumnDescriptor;
54 import org.apache.hadoop.hbase.HConstants;
55 import org.apache.hadoop.hbase.HRegionInfo;
56 import org.apache.hadoop.hbase.HRegionLocation;
57 import org.apache.hadoop.hbase.HTableDescriptor;
58 import org.apache.hadoop.hbase.KeyValue;
59 import org.apache.hadoop.hbase.ServerName;
60 import org.apache.hadoop.hbase.TableName;
61 import org.apache.hadoop.hbase.TableNotFoundException;
62 import org.apache.hadoop.hbase.classification.InterfaceAudience;
63 import org.apache.hadoop.hbase.client.Admin;
64 import org.apache.hadoop.hbase.client.Append;
65 import org.apache.hadoop.hbase.client.Delete;
66 import org.apache.hadoop.hbase.client.Durability;
67 import org.apache.hadoop.hbase.client.Get;
68 import org.apache.hadoop.hbase.client.HBaseAdmin;
69 import org.apache.hadoop.hbase.client.Increment;
70 import org.apache.hadoop.hbase.client.OperationWithAttributes;
71 import org.apache.hadoop.hbase.client.Put;
72 import org.apache.hadoop.hbase.client.RegionLocator;
73 import org.apache.hadoop.hbase.client.Result;
74 import org.apache.hadoop.hbase.client.ResultScanner;
75 import org.apache.hadoop.hbase.client.Scan;
76 import org.apache.hadoop.hbase.client.Table;
77 import org.apache.hadoop.hbase.filter.Filter;
78 import org.apache.hadoop.hbase.filter.ParseFilter;
79 import org.apache.hadoop.hbase.filter.PrefixFilter;
80 import org.apache.hadoop.hbase.filter.WhileMatchFilter;
81 import org.apache.hadoop.hbase.jetty.SslSelectChannelConnectorSecure;
82 import org.apache.hadoop.hbase.security.SaslUtil;
83 import org.apache.hadoop.hbase.security.SaslUtil.QualityOfProtection;
84 import org.apache.hadoop.hbase.security.SecurityUtil;
85 import org.apache.hadoop.hbase.security.UserProvider;
86 import org.apache.hadoop.hbase.thrift.CallQueue.Call;
87 import org.apache.hadoop.hbase.thrift.generated.AlreadyExists;
88 import org.apache.hadoop.hbase.thrift.generated.BatchMutation;
89 import org.apache.hadoop.hbase.thrift.generated.ColumnDescriptor;
90 import org.apache.hadoop.hbase.thrift.generated.Hbase;
91 import org.apache.hadoop.hbase.thrift.generated.IOError;
92 import org.apache.hadoop.hbase.thrift.generated.IllegalArgument;
93 import org.apache.hadoop.hbase.thrift.generated.Mutation;
94 import org.apache.hadoop.hbase.thrift.generated.TAppend;
95 import org.apache.hadoop.hbase.thrift.generated.TCell;
96 import org.apache.hadoop.hbase.thrift.generated.TIncrement;
97 import org.apache.hadoop.hbase.thrift.generated.TRegionInfo;
98 import org.apache.hadoop.hbase.thrift.generated.TRowResult;
99 import org.apache.hadoop.hbase.thrift.generated.TScan;
100 import org.apache.hadoop.hbase.util.Bytes;
101 import org.apache.hadoop.hbase.util.ConnectionCache;
102 import org.apache.hadoop.hbase.util.DNS;
103 import org.apache.hadoop.hbase.util.HttpServerUtil;
104 import org.apache.hadoop.hbase.util.JvmPauseMonitor;
105 import org.apache.hadoop.hbase.util.Strings;
106 import org.apache.hadoop.security.SaslRpcServer.SaslGssCallbackHandler;
107 import org.apache.hadoop.security.UserGroupInformation;
108 import org.apache.hadoop.security.authorize.ProxyUsers;
109 import org.apache.thrift.TException;
110 import org.apache.thrift.TProcessor;
111 import org.apache.thrift.protocol.TBinaryProtocol;
112 import org.apache.thrift.protocol.TCompactProtocol;
113 import org.apache.thrift.protocol.TProtocol;
114 import org.apache.thrift.protocol.TProtocolFactory;
115 import org.apache.thrift.server.THsHaServer;
116 import org.apache.thrift.server.TNonblockingServer;
117 import org.apache.thrift.server.TServer;
118 import org.apache.thrift.server.TServlet;
119 import org.apache.thrift.server.TThreadedSelectorServer;
120 import org.apache.thrift.transport.TFramedTransport;
121 import org.apache.thrift.transport.TNonblockingServerSocket;
122 import org.apache.thrift.transport.TNonblockingServerTransport;
123 import org.apache.thrift.transport.TSaslServerTransport;
124 import org.apache.thrift.transport.TServerSocket;
125 import org.apache.thrift.transport.TServerTransport;
126 import org.apache.thrift.transport.TTransportFactory;
127 import org.mortbay.jetty.Connector;
128 import org.mortbay.jetty.Server;
129 import org.mortbay.jetty.nio.SelectChannelConnector;
130 import org.mortbay.jetty.servlet.Context;
131 import org.mortbay.jetty.servlet.ServletHolder;
132 import org.mortbay.jetty.webapp.WebAppContext;
133 import org.mortbay.thread.QueuedThreadPool;
134
135 import com.google.common.base.Joiner;
136 import com.google.common.base.Throwables;
137 import com.google.common.util.concurrent.ThreadFactoryBuilder;
138
139
140
141
142
143 @InterfaceAudience.Private
144 public class ThriftServerRunner implements Runnable {
145
146 private static final Log LOG = LogFactory.getLog(ThriftServerRunner.class);
147
148 static final String SERVER_TYPE_CONF_KEY =
149 "hbase.regionserver.thrift.server.type";
150
151 static final String BIND_CONF_KEY = "hbase.regionserver.thrift.ipaddress";
152 static final String COMPACT_CONF_KEY = "hbase.regionserver.thrift.compact";
153 static final String FRAMED_CONF_KEY = "hbase.regionserver.thrift.framed";
154 static final String MAX_FRAME_SIZE_CONF_KEY = "hbase.regionserver.thrift.framed.max_frame_size_in_mb";
155 static final String PORT_CONF_KEY = "hbase.regionserver.thrift.port";
156 static final String COALESCE_INC_KEY = "hbase.regionserver.thrift.coalesceIncrement";
157 static final String USE_HTTP_CONF_KEY = "hbase.regionserver.thrift.http";
158 static final String HTTP_MIN_THREADS = "hbase.thrift.http_threads.min";
159 static final String HTTP_MAX_THREADS = "hbase.thrift.http_threads.max";
160
161 static final String THRIFT_SSL_ENABLED = "hbase.thrift.ssl.enabled";
162 static final String THRIFT_SSL_KEYSTORE_STORE = "hbase.thrift.ssl.keystore.store";
163 static final String THRIFT_SSL_KEYSTORE_PASSWORD = "hbase.thrift.ssl.keystore.password";
164 static final String THRIFT_SSL_KEYSTORE_KEYPASSWORD = "hbase.thrift.ssl.keystore.keypassword";
165
166
167
168
169
170
171 public static final String THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY =
172 "hbase.thrift.server.socket.read.timeout";
173 public static final int THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT = 60000;
174
175
176
177
178
179
180
181
182
183
184
185 static final String THRIFT_QOP_KEY = "hbase.thrift.security.qop";
186 static final String BACKLOG_CONF_KEY = "hbase.regionserver.thrift.backlog";
187
188 private static final String DEFAULT_BIND_ADDR = "0.0.0.0";
189 public static final int DEFAULT_LISTEN_PORT = 9090;
190 public static final int HREGION_VERSION = 1;
191 static final String THRIFT_SUPPORT_PROXYUSER = "hbase.thrift.support.proxyuser";
192 private final int listenPort;
193
194 private Configuration conf;
195 volatile TServer tserver;
196 volatile Server httpServer;
197 private final Hbase.Iface handler;
198 private final ThriftMetrics metrics;
199 private final HBaseHandler hbaseHandler;
200 private final UserGroupInformation realUser;
201
202 private SaslUtil.QualityOfProtection qop;
203 private String host;
204
205 private final boolean securityEnabled;
206 private final boolean doAsEnabled;
207
208 private final JvmPauseMonitor pauseMonitor;
209
210 static String THRIFT_HTTP_ALLOW_OPTIONS_METHOD = "hbase.thrift.http.allow.options.method";
211 private static boolean THRIFT_HTTP_ALLOW_OPTIONS_METHOD_DEFAULT = false;
212
213
214 enum ImplType {
215 HS_HA("hsha", true, THsHaServer.class, true),
216 NONBLOCKING("nonblocking", true, TNonblockingServer.class, true),
217 THREAD_POOL("threadpool", false, TBoundedThreadPoolServer.class, true),
218 THREADED_SELECTOR(
219 "threadedselector", true, TThreadedSelectorServer.class, true);
220
221 public static final ImplType DEFAULT = THREAD_POOL;
222
223 final String option;
224 final boolean isAlwaysFramed;
225 final Class<? extends TServer> serverClass;
226 final boolean canSpecifyBindIP;
227
228 ImplType(String option, boolean isAlwaysFramed,
229 Class<? extends TServer> serverClass, boolean canSpecifyBindIP) {
230 this.option = option;
231 this.isAlwaysFramed = isAlwaysFramed;
232 this.serverClass = serverClass;
233 this.canSpecifyBindIP = canSpecifyBindIP;
234 }
235
236
237
238
239
240 @Override
241 public String toString() {
242 return "-" + option;
243 }
244
245 String getDescription() {
246 StringBuilder sb = new StringBuilder("Use the " +
247 serverClass.getSimpleName());
248 if (isAlwaysFramed) {
249 sb.append(" This implies the framed transport.");
250 }
251 if (this == DEFAULT) {
252 sb.append("This is the default.");
253 }
254 return sb.toString();
255 }
256
257 static OptionGroup createOptionGroup() {
258 OptionGroup group = new OptionGroup();
259 for (ImplType t : values()) {
260 group.addOption(new Option(t.option, t.getDescription()));
261 }
262 return group;
263 }
264
265 static ImplType getServerImpl(Configuration conf) {
266 String confType = conf.get(SERVER_TYPE_CONF_KEY, THREAD_POOL.option);
267 for (ImplType t : values()) {
268 if (confType.equals(t.option)) {
269 return t;
270 }
271 }
272 throw new AssertionError("Unknown server ImplType.option:" + confType);
273 }
274
275 static void setServerImpl(CommandLine cmd, Configuration conf) {
276 ImplType chosenType = null;
277 int numChosen = 0;
278 for (ImplType t : values()) {
279 if (cmd.hasOption(t.option)) {
280 chosenType = t;
281 ++numChosen;
282 }
283 }
284 if (numChosen < 1) {
285 LOG.info("Using default thrift server type");
286 chosenType = DEFAULT;
287 } else if (numChosen > 1) {
288 throw new AssertionError("Exactly one option out of " +
289 Arrays.toString(values()) + " has to be specified");
290 }
291 LOG.info("Using thrift server type " + chosenType.option);
292 conf.set(SERVER_TYPE_CONF_KEY, chosenType.option);
293 }
294
295 public String simpleClassName() {
296 return serverClass.getSimpleName();
297 }
298
299 public static List<String> serversThatCannotSpecifyBindIP() {
300 List<String> l = new ArrayList<String>();
301 for (ImplType t : values()) {
302 if (!t.canSpecifyBindIP) {
303 l.add(t.simpleClassName());
304 }
305 }
306 return l;
307 }
308
309 }
310
311 public ThriftServerRunner(Configuration conf) throws IOException {
312 UserProvider userProvider = UserProvider.instantiate(conf);
313
314 securityEnabled = userProvider.isHadoopSecurityEnabled()
315 && userProvider.isHBaseSecurityEnabled();
316 if (securityEnabled) {
317 host = Strings.domainNamePointerToHostName(DNS.getDefaultHost(
318 conf.get("hbase.thrift.dns.interface", "default"),
319 conf.get("hbase.thrift.dns.nameserver", "default")));
320 userProvider.login("hbase.thrift.keytab.file",
321 "hbase.thrift.kerberos.principal", host);
322 }
323 this.conf = HBaseConfiguration.create(conf);
324 this.listenPort = conf.getInt(PORT_CONF_KEY, DEFAULT_LISTEN_PORT);
325 this.metrics = new ThriftMetrics(conf, ThriftMetrics.ThriftServerType.ONE);
326 this.pauseMonitor = new JvmPauseMonitor(conf, this.metrics.getSource());
327 this.hbaseHandler = new HBaseHandler(conf, userProvider);
328 this.hbaseHandler.initMetrics(metrics);
329 this.handler = HbaseHandlerMetricsProxy.newInstance(
330 hbaseHandler, metrics, conf);
331 this.realUser = userProvider.getCurrent().getUGI();
332 String strQop = conf.get(THRIFT_QOP_KEY);
333 if (strQop != null) {
334 this.qop = SaslUtil.getQop(strQop);
335 }
336 doAsEnabled = conf.getBoolean(THRIFT_SUPPORT_PROXYUSER, false);
337 if (doAsEnabled) {
338 if (!conf.getBoolean(USE_HTTP_CONF_KEY, false)) {
339 LOG.warn("Fail to enable the doAs feature. hbase.regionserver.thrift.http is not configured ");
340 }
341 }
342 if (qop != null) {
343 if (qop != QualityOfProtection.AUTHENTICATION &&
344 qop != QualityOfProtection.INTEGRITY &&
345 qop != QualityOfProtection.PRIVACY) {
346 throw new IOException(String.format("Invalide %s: It must be one of %s, %s, or %s.",
347 THRIFT_QOP_KEY,
348 QualityOfProtection.AUTHENTICATION.name(),
349 QualityOfProtection.INTEGRITY.name(),
350 QualityOfProtection.PRIVACY.name()));
351 }
352 checkHttpSecurity(qop, conf);
353 if (!securityEnabled) {
354 throw new IOException("Thrift server must"
355 + " run in secure mode to support authentication");
356 }
357 }
358 }
359
360 private void checkHttpSecurity(QualityOfProtection qop, Configuration conf) {
361 if (qop == QualityOfProtection.PRIVACY &&
362 conf.getBoolean(USE_HTTP_CONF_KEY, false) &&
363 !conf.getBoolean(THRIFT_SSL_ENABLED, false)) {
364 throw new IllegalArgumentException("Thrift HTTP Server's QoP is privacy, but " +
365 THRIFT_SSL_ENABLED + " is false");
366 }
367 }
368
369
370
371
372 @Override
373 public void run() {
374 realUser.doAs(new PrivilegedAction<Object>() {
375 @Override
376 public Object run() {
377 try {
378 pauseMonitor.start();
379 if (conf.getBoolean(USE_HTTP_CONF_KEY, false)) {
380 setupHTTPServer();
381 httpServer.start();
382 httpServer.join();
383 } else {
384 setupServer();
385 tserver.serve();
386 }
387 } catch (Exception e) {
388 LOG.fatal("Cannot run ThriftServer", e);
389
390 System.exit(-1);
391 }
392 return null;
393 }
394 });
395
396 }
397
398 public void shutdown() {
399 if (pauseMonitor != null) {
400 pauseMonitor.stop();
401 }
402 if (tserver != null) {
403 tserver.stop();
404 tserver = null;
405 }
406 if (httpServer != null) {
407 try {
408 httpServer.stop();
409 httpServer = null;
410 } catch (Exception e) {
411 LOG.error("Problem encountered in shutting down HTTP server " + e.getCause());
412 }
413 httpServer = null;
414 }
415 }
416
417 private void setupHTTPServer() throws IOException {
418 TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
419 TProcessor processor = new Hbase.Processor<Hbase.Iface>(handler);
420 TServlet thriftHttpServlet = new ThriftHttpServlet(processor, protocolFactory, realUser,
421 conf, hbaseHandler, securityEnabled, doAsEnabled);
422
423 httpServer = new Server();
424
425 Context context = new WebAppContext();
426 context.setContextPath("/");
427 context.setResourceBase("hbase-webapps/");
428 String httpPath = "/*";
429 httpServer.setHandler(context);
430 context.addServlet(new ServletHolder(thriftHttpServlet), httpPath);
431 HttpServerUtil.constrainHttpMethods(context,
432 conf.getBoolean(THRIFT_HTTP_ALLOW_OPTIONS_METHOD, THRIFT_HTTP_ALLOW_OPTIONS_METHOD_DEFAULT));
433
434
435 Connector connector = new SelectChannelConnector();
436 if(conf.getBoolean(THRIFT_SSL_ENABLED, false)) {
437 SslSelectChannelConnectorSecure sslConnector = new SslSelectChannelConnectorSecure();
438 String keystore = conf.get(THRIFT_SSL_KEYSTORE_STORE);
439 String password = HBaseConfiguration.getPassword(conf,
440 THRIFT_SSL_KEYSTORE_PASSWORD, null);
441 String keyPassword = HBaseConfiguration.getPassword(conf,
442 THRIFT_SSL_KEYSTORE_KEYPASSWORD, password);
443 sslConnector.setKeystore(keystore);
444 sslConnector.setPassword(password);
445 sslConnector.setKeyPassword(keyPassword);
446 connector = sslConnector;
447 }
448 String host = getBindAddress(conf).getHostAddress();
449 connector.setPort(listenPort);
450 connector.setHost(host);
451 connector.setHeaderBufferSize(1024 * 64);
452 httpServer.addConnector(connector);
453
454 if (doAsEnabled) {
455 ProxyUsers.refreshSuperUserGroupsConfiguration(conf);
456 }
457
458
459
460
461
462
463 int minThreads = conf.getInt(HTTP_MIN_THREADS, 2);
464 int maxThreads = conf.getInt(HTTP_MAX_THREADS, 100);
465 QueuedThreadPool threadPool = new QueuedThreadPool(maxThreads);
466 threadPool.setMinThreads(minThreads);
467 httpServer.setThreadPool(threadPool);
468
469 httpServer.setSendServerVersion(false);
470 httpServer.setSendDateHeader(false);
471 httpServer.setStopAtShutdown(true);
472
473 LOG.info("Starting Thrift HTTP Server on " + Integer.toString(listenPort));
474 }
475
476
477
478
479 private void setupServer() throws Exception {
480
481 TProtocolFactory protocolFactory;
482 if (conf.getBoolean(COMPACT_CONF_KEY, false)) {
483 LOG.debug("Using compact protocol");
484 protocolFactory = new TCompactProtocol.Factory();
485 } else {
486 LOG.debug("Using binary protocol");
487 protocolFactory = new TBinaryProtocol.Factory();
488 }
489
490 final TProcessor p = new Hbase.Processor<Hbase.Iface>(handler);
491 ImplType implType = ImplType.getServerImpl(conf);
492 TProcessor processor = p;
493
494
495 TTransportFactory transportFactory;
496 if (conf.getBoolean(FRAMED_CONF_KEY, false) || implType.isAlwaysFramed) {
497 if (qop != null) {
498 throw new RuntimeException("Thrift server authentication"
499 + " doesn't work with framed transport yet");
500 }
501 transportFactory = new TFramedTransport.Factory(
502 conf.getInt(MAX_FRAME_SIZE_CONF_KEY, 2) * 1024 * 1024);
503 LOG.debug("Using framed transport");
504 } else if (qop == null) {
505 transportFactory = new TTransportFactory();
506 } else {
507
508 String name = SecurityUtil.getUserFromPrincipal(
509 conf.get("hbase.thrift.kerberos.principal"));
510 Map<String, String> saslProperties = SaslUtil.initSaslProperties(qop.name());
511 TSaslServerTransport.Factory saslFactory = new TSaslServerTransport.Factory();
512 saslFactory.addServerDefinition("GSSAPI", name, host, saslProperties,
513 new SaslGssCallbackHandler() {
514 @Override
515 public void handle(Callback[] callbacks)
516 throws UnsupportedCallbackException {
517 AuthorizeCallback ac = null;
518 for (Callback callback : callbacks) {
519 if (callback instanceof AuthorizeCallback) {
520 ac = (AuthorizeCallback) callback;
521 } else {
522 throw new UnsupportedCallbackException(callback,
523 "Unrecognized SASL GSSAPI Callback");
524 }
525 }
526 if (ac != null) {
527 String authid = ac.getAuthenticationID();
528 String authzid = ac.getAuthorizationID();
529 if (!authid.equals(authzid)) {
530 ac.setAuthorized(false);
531 } else {
532 ac.setAuthorized(true);
533 String userName = SecurityUtil.getUserFromPrincipal(authzid);
534 LOG.info("Effective user: " + userName);
535 ac.setAuthorizedID(userName);
536 }
537 }
538 }
539 });
540 transportFactory = saslFactory;
541
542
543 processor = new TProcessor() {
544 @Override
545 public boolean process(TProtocol inProt,
546 TProtocol outProt) throws TException {
547 TSaslServerTransport saslServerTransport =
548 (TSaslServerTransport)inProt.getTransport();
549 SaslServer saslServer = saslServerTransport.getSaslServer();
550 String principal = saslServer.getAuthorizationID();
551 hbaseHandler.setEffectiveUser(principal);
552 return p.process(inProt, outProt);
553 }
554 };
555 }
556
557 if (conf.get(BIND_CONF_KEY) != null && !implType.canSpecifyBindIP) {
558 LOG.error("Server types " + Joiner.on(", ").join(
559 ImplType.serversThatCannotSpecifyBindIP()) + " don't support IP " +
560 "address binding at the moment. See " +
561 "https://issues.apache.org/jira/browse/HBASE-2155 for details.");
562 throw new RuntimeException(
563 "-" + BIND_CONF_KEY + " not supported with " + implType);
564 }
565
566
567 int backlog = conf.getInt(BACKLOG_CONF_KEY, 0);
568
569 if (implType == ImplType.HS_HA || implType == ImplType.NONBLOCKING ||
570 implType == ImplType.THREADED_SELECTOR) {
571 InetAddress listenAddress = getBindAddress(conf);
572 TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(
573 new InetSocketAddress(listenAddress, listenPort));
574
575 if (implType == ImplType.NONBLOCKING) {
576 TNonblockingServer.Args serverArgs =
577 new TNonblockingServer.Args(serverTransport);
578 serverArgs.processor(processor)
579 .transportFactory(transportFactory)
580 .protocolFactory(protocolFactory);
581 tserver = new TNonblockingServer(serverArgs);
582 } else if (implType == ImplType.HS_HA) {
583 THsHaServer.Args serverArgs = new THsHaServer.Args(serverTransport);
584 CallQueue callQueue =
585 new CallQueue(new LinkedBlockingQueue<Call>(), metrics);
586 ExecutorService executorService = createExecutor(
587 callQueue, serverArgs.getMinWorkerThreads(), serverArgs.getMaxWorkerThreads());
588 serverArgs.executorService(executorService)
589 .processor(processor)
590 .transportFactory(transportFactory)
591 .protocolFactory(protocolFactory);
592 tserver = new THsHaServer(serverArgs);
593 } else {
594 TThreadedSelectorServer.Args serverArgs =
595 new HThreadedSelectorServerArgs(serverTransport, conf);
596 CallQueue callQueue =
597 new CallQueue(new LinkedBlockingQueue<Call>(), metrics);
598 ExecutorService executorService = createExecutor(
599 callQueue, serverArgs.getWorkerThreads(), serverArgs.getWorkerThreads());
600 serverArgs.executorService(executorService)
601 .processor(processor)
602 .transportFactory(transportFactory)
603 .protocolFactory(protocolFactory);
604 tserver = new TThreadedSelectorServer(serverArgs);
605 }
606 LOG.info("starting HBase " + implType.simpleClassName() +
607 " server on " + Integer.toString(listenPort));
608 } else if (implType == ImplType.THREAD_POOL) {
609
610 InetAddress listenAddress = getBindAddress(conf);
611 int readTimeout = conf.getInt(THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY,
612 THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT);
613 TServerTransport serverTransport = new TServerSocket(
614 new TServerSocket.ServerSocketTransportArgs().
615 bindAddr(new InetSocketAddress(listenAddress, listenPort)).
616 backlog(backlog).
617 clientTimeout(readTimeout));
618
619 TBoundedThreadPoolServer.Args serverArgs =
620 new TBoundedThreadPoolServer.Args(serverTransport, conf);
621 serverArgs.processor(processor)
622 .transportFactory(transportFactory)
623 .protocolFactory(protocolFactory);
624 LOG.info("starting " + ImplType.THREAD_POOL.simpleClassName() + " on "
625 + listenAddress + ":" + Integer.toString(listenPort)
626 + " with readTimeout " + readTimeout + "ms; " + serverArgs);
627 TBoundedThreadPoolServer tserver =
628 new TBoundedThreadPoolServer(serverArgs, metrics);
629 this.tserver = tserver;
630 } else {
631 throw new AssertionError("Unsupported Thrift server implementation: " +
632 implType.simpleClassName());
633 }
634
635
636 if (tserver.getClass() != implType.serverClass) {
637 throw new AssertionError("Expected to create Thrift server class " +
638 implType.serverClass.getName() + " but got " +
639 tserver.getClass().getName());
640 }
641
642
643
644 registerFilters(conf);
645 }
646
647 ExecutorService createExecutor(BlockingQueue<Runnable> callQueue,
648 int minWorkers, int maxWorkers) {
649 ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
650 tfb.setDaemon(true);
651 tfb.setNameFormat("thrift-worker-%d");
652 return new THBaseThreadPoolExecutor(minWorkers, maxWorkers,
653 Long.MAX_VALUE, TimeUnit.SECONDS, callQueue, tfb.build(), metrics);
654 }
655
656 private InetAddress getBindAddress(Configuration conf)
657 throws UnknownHostException {
658 String bindAddressStr = conf.get(BIND_CONF_KEY, DEFAULT_BIND_ADDR);
659 return InetAddress.getByName(bindAddressStr);
660 }
661
662 protected static class ResultScannerWrapper {
663
664 private final ResultScanner scanner;
665 private final boolean sortColumns;
666 public ResultScannerWrapper(ResultScanner resultScanner,
667 boolean sortResultColumns) {
668 scanner = resultScanner;
669 sortColumns = sortResultColumns;
670 }
671
672 public ResultScanner getScanner() {
673 return scanner;
674 }
675
676 public boolean isColumnSorted() {
677 return sortColumns;
678 }
679 }
680
681
682
683
684
685 public static class HBaseHandler implements Hbase.Iface {
686 protected Configuration conf;
687 protected static final Log LOG = LogFactory.getLog(HBaseHandler.class);
688
689
690 protected int nextScannerId = 0;
691 protected HashMap<Integer, ResultScannerWrapper> scannerMap = null;
692 private ThriftMetrics metrics = null;
693
694 private final ConnectionCache connectionCache;
695 IncrementCoalescer coalescer = null;
696
697 static final String CLEANUP_INTERVAL = "hbase.thrift.connection.cleanup-interval";
698 static final String MAX_IDLETIME = "hbase.thrift.connection.max-idletime";
699
700
701
702
703
704
705
706 byte[][] getAllColumns(Table table) throws IOException {
707 HColumnDescriptor[] cds = table.getTableDescriptor().getColumnFamilies();
708 byte[][] columns = new byte[cds.length][];
709 for (int i = 0; i < cds.length; i++) {
710 columns[i] = Bytes.add(cds[i].getName(),
711 KeyValue.COLUMN_FAMILY_DELIM_ARRAY);
712 }
713 return columns;
714 }
715
716
717
718
719
720
721
722
723
724 public Table getTable(final byte[] tableName) throws
725 IOException {
726 String table = Bytes.toString(tableName);
727 return connectionCache.getTable(table);
728 }
729
730 public Table getTable(final ByteBuffer tableName) throws IOException {
731 return getTable(getBytes(tableName));
732 }
733
734
735
736
737
738
739
740
741 protected synchronized int addScanner(ResultScanner scanner,boolean sortColumns) {
742 int id = nextScannerId++;
743 ResultScannerWrapper resultScannerWrapper = new ResultScannerWrapper(scanner, sortColumns);
744 scannerMap.put(id, resultScannerWrapper);
745 return id;
746 }
747
748
749
750
751
752
753
754 protected synchronized ResultScannerWrapper getScanner(int id) {
755 return scannerMap.get(id);
756 }
757
758
759
760
761
762
763
764
765 protected synchronized ResultScannerWrapper removeScanner(int id) {
766 return scannerMap.remove(id);
767 }
768
769 protected HBaseHandler(final Configuration c,
770 final UserProvider userProvider) throws IOException {
771 this.conf = c;
772 scannerMap = new HashMap<Integer, ResultScannerWrapper>();
773 this.coalescer = new IncrementCoalescer(this);
774
775 int cleanInterval = conf.getInt(CLEANUP_INTERVAL, 10 * 1000);
776 int maxIdleTime = conf.getInt(MAX_IDLETIME, 10 * 60 * 1000);
777 connectionCache = new ConnectionCache(
778 conf, userProvider, cleanInterval, maxIdleTime);
779 }
780
781
782
783
784 private Admin getAdmin() throws IOException {
785 return connectionCache.getAdmin();
786 }
787
788 void setEffectiveUser(String effectiveUser) {
789 connectionCache.setEffectiveUser(effectiveUser);
790 }
791
792 @Override
793 public void enableTable(ByteBuffer tableName) throws IOError {
794 try{
795 getAdmin().enableTable(getTableName(tableName));
796 } catch (IOException e) {
797 LOG.warn(e.getMessage(), e);
798 throw getIOError(e);
799 }
800 }
801
802 @Override
803 public void disableTable(ByteBuffer tableName) throws IOError{
804 try{
805 getAdmin().disableTable(getTableName(tableName));
806 } catch (IOException e) {
807 LOG.warn(e.getMessage(), e);
808 throw getIOError(e);
809 }
810 }
811
812 @Override
813 public boolean isTableEnabled(ByteBuffer tableName) throws IOError {
814 try {
815 return this.connectionCache.getAdmin().isTableEnabled(getTableName(tableName));
816 } catch (IOException e) {
817 LOG.warn(e.getMessage(), e);
818 throw getIOError(e);
819 }
820 }
821
822 @Override
823 public void compact(ByteBuffer tableNameOrRegionName) throws IOError {
824 try {
825
826
827
828 ((HBaseAdmin) getAdmin()).compact(getBytes(tableNameOrRegionName));
829 } catch (IOException e) {
830 LOG.warn(e.getMessage(), e);
831 throw getIOError(e);
832 }
833 }
834
835 @Override
836 public void majorCompact(ByteBuffer tableNameOrRegionName) throws IOError {
837 try {
838
839
840
841 ((HBaseAdmin) getAdmin()).majorCompact(getBytes(tableNameOrRegionName));
842 } catch (IOException e) {
843 LOG.warn(e.getMessage(), e);
844 throw getIOError(e);
845 }
846 }
847
848 @Override
849 public List<ByteBuffer> getTableNames() throws IOError {
850 try {
851 TableName[] tableNames = this.getAdmin().listTableNames();
852 ArrayList<ByteBuffer> list = new ArrayList<ByteBuffer>(tableNames.length);
853 for (int i = 0; i < tableNames.length; i++) {
854 list.add(ByteBuffer.wrap(tableNames[i].getName()));
855 }
856 return list;
857 } catch (IOException e) {
858 LOG.warn(e.getMessage(), e);
859 throw getIOError(e);
860 }
861 }
862
863
864
865
866 @Override
867 public List<TRegionInfo> getTableRegions(ByteBuffer tableName)
868 throws IOError {
869 try (RegionLocator locator = connectionCache.getRegionLocator(getBytes(tableName))) {
870 List<HRegionLocation> regionLocations = locator.getAllRegionLocations();
871 List<TRegionInfo> results = new ArrayList<TRegionInfo>();
872 for (HRegionLocation regionLocation : regionLocations) {
873 HRegionInfo info = regionLocation.getRegionInfo();
874 ServerName serverName = regionLocation.getServerName();
875 TRegionInfo region = new TRegionInfo();
876 region.serverName = ByteBuffer.wrap(
877 Bytes.toBytes(serverName.getHostname()));
878 region.port = serverName.getPort();
879 region.startKey = ByteBuffer.wrap(info.getStartKey());
880 region.endKey = ByteBuffer.wrap(info.getEndKey());
881 region.id = info.getRegionId();
882 region.name = ByteBuffer.wrap(info.getRegionName());
883 region.version = info.getVersion();
884 results.add(region);
885 }
886 return results;
887 } catch (TableNotFoundException e) {
888
889 return Collections.emptyList();
890 } catch (IOException e){
891 LOG.warn(e.getMessage(), e);
892 throw getIOError(e);
893 }
894 }
895
896 @Override
897 public List<TCell> get(
898 ByteBuffer tableName, ByteBuffer row, ByteBuffer column,
899 Map<ByteBuffer, ByteBuffer> attributes)
900 throws IOError {
901 byte [][] famAndQf = KeyValue.parseColumn(getBytes(column));
902 if (famAndQf.length == 1) {
903 return get(tableName, row, famAndQf[0], null, attributes);
904 }
905 if (famAndQf.length == 2) {
906 return get(tableName, row, famAndQf[0], famAndQf[1], attributes);
907 }
908 throw new IllegalArgumentException("Invalid familyAndQualifier provided.");
909 }
910
911
912
913
914
915
916
917
918 protected List<TCell> get(ByteBuffer tableName,
919 ByteBuffer row,
920 byte[] family,
921 byte[] qualifier,
922 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
923 Table table = null;
924 try {
925 table = getTable(tableName);
926 Get get = new Get(getBytes(row));
927 addAttributes(get, attributes);
928 if (qualifier == null) {
929 get.addFamily(family);
930 } else {
931 get.addColumn(family, qualifier);
932 }
933 Result result = table.get(get);
934 return ThriftUtilities.cellFromHBase(result.rawCells());
935 } catch (IOException e) {
936 LOG.warn(e.getMessage(), e);
937 throw getIOError(e);
938 } finally {
939 closeTable(table);
940 }
941 }
942
943 @Override
944 public List<TCell> getVer(ByteBuffer tableName, ByteBuffer row, ByteBuffer column,
945 int numVersions, Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
946 byte [][] famAndQf = KeyValue.parseColumn(getBytes(column));
947 if(famAndQf.length == 1) {
948 return getVer(tableName, row, famAndQf[0], null, numVersions, attributes);
949 }
950 if (famAndQf.length == 2) {
951 return getVer(tableName, row, famAndQf[0], famAndQf[1], numVersions, attributes);
952 }
953 throw new IllegalArgumentException("Invalid familyAndQualifier provided.");
954
955 }
956
957
958
959
960
961
962
963
964
965 public List<TCell> getVer(ByteBuffer tableName, ByteBuffer row, byte[] family,
966 byte[] qualifier, int numVersions, Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
967
968 Table table = null;
969 try {
970 table = getTable(tableName);
971 Get get = new Get(getBytes(row));
972 addAttributes(get, attributes);
973 if (null == qualifier) {
974 get.addFamily(family);
975 } else {
976 get.addColumn(family, qualifier);
977 }
978 get.setMaxVersions(numVersions);
979 Result result = table.get(get);
980 return ThriftUtilities.cellFromHBase(result.rawCells());
981 } catch (IOException e) {
982 LOG.warn(e.getMessage(), e);
983 throw getIOError(e);
984 } finally{
985 closeTable(table);
986 }
987 }
988
989 @Override
990 public List<TCell> getVerTs(ByteBuffer tableName, ByteBuffer row, ByteBuffer column,
991 long timestamp, int numVersions, Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
992 byte [][] famAndQf = KeyValue.parseColumn(getBytes(column));
993 if (famAndQf.length == 1) {
994 return getVerTs(tableName, row, famAndQf[0], null, timestamp, numVersions, attributes);
995 }
996 if (famAndQf.length == 2) {
997 return getVerTs(tableName, row, famAndQf[0], famAndQf[1], timestamp, numVersions,
998 attributes);
999 }
1000 throw new IllegalArgumentException("Invalid familyAndQualifier provided.");
1001 }
1002
1003
1004
1005
1006
1007
1008
1009
1010 protected List<TCell> getVerTs(ByteBuffer tableName, ByteBuffer row, byte[] family,
1011 byte[] qualifier, long timestamp, int numVersions, Map<ByteBuffer, ByteBuffer> attributes)
1012 throws IOError {
1013
1014 Table table = null;
1015 try {
1016 table = getTable(tableName);
1017 Get get = new Get(getBytes(row));
1018 addAttributes(get, attributes);
1019 if (null == qualifier) {
1020 get.addFamily(family);
1021 } else {
1022 get.addColumn(family, qualifier);
1023 }
1024 get.setTimeRange(0, timestamp);
1025 get.setMaxVersions(numVersions);
1026 Result result = table.get(get);
1027 return ThriftUtilities.cellFromHBase(result.rawCells());
1028 } catch (IOException e) {
1029 LOG.warn(e.getMessage(), e);
1030 throw getIOError(e);
1031 } finally{
1032 closeTable(table);
1033 }
1034 }
1035
1036 @Override
1037 public List<TRowResult> getRow(ByteBuffer tableName, ByteBuffer row,
1038 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1039 return getRowWithColumnsTs(tableName, row, null,
1040 HConstants.LATEST_TIMESTAMP,
1041 attributes);
1042 }
1043
1044 @Override
1045 public List<TRowResult> getRowWithColumns(ByteBuffer tableName,
1046 ByteBuffer row,
1047 List<ByteBuffer> columns,
1048 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1049 return getRowWithColumnsTs(tableName, row, columns,
1050 HConstants.LATEST_TIMESTAMP,
1051 attributes);
1052 }
1053
1054 @Override
1055 public List<TRowResult> getRowTs(ByteBuffer tableName, ByteBuffer row,
1056 long timestamp, Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1057 return getRowWithColumnsTs(tableName, row, null,
1058 timestamp, attributes);
1059 }
1060
1061 @Override
1062 public List<TRowResult> getRowWithColumnsTs(
1063 ByteBuffer tableName, ByteBuffer row, List<ByteBuffer> columns,
1064 long timestamp, Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1065
1066 Table table = null;
1067 try {
1068 table = getTable(tableName);
1069 if (columns == null) {
1070 Get get = new Get(getBytes(row));
1071 addAttributes(get, attributes);
1072 get.setTimeRange(0, timestamp);
1073 Result result = table.get(get);
1074 return ThriftUtilities.rowResultFromHBase(result);
1075 }
1076 Get get = new Get(getBytes(row));
1077 addAttributes(get, attributes);
1078 for(ByteBuffer column : columns) {
1079 byte [][] famAndQf = KeyValue.parseColumn(getBytes(column));
1080 if (famAndQf.length == 1) {
1081 get.addFamily(famAndQf[0]);
1082 } else {
1083 get.addColumn(famAndQf[0], famAndQf[1]);
1084 }
1085 }
1086 get.setTimeRange(0, timestamp);
1087 Result result = table.get(get);
1088 return ThriftUtilities.rowResultFromHBase(result);
1089 } catch (IOException e) {
1090 LOG.warn(e.getMessage(), e);
1091 throw getIOError(e);
1092 } finally{
1093 closeTable(table);
1094 }
1095 }
1096
1097 @Override
1098 public List<TRowResult> getRows(ByteBuffer tableName,
1099 List<ByteBuffer> rows,
1100 Map<ByteBuffer, ByteBuffer> attributes)
1101 throws IOError {
1102 return getRowsWithColumnsTs(tableName, rows, null,
1103 HConstants.LATEST_TIMESTAMP,
1104 attributes);
1105 }
1106
1107 @Override
1108 public List<TRowResult> getRowsWithColumns(ByteBuffer tableName,
1109 List<ByteBuffer> rows,
1110 List<ByteBuffer> columns,
1111 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1112 return getRowsWithColumnsTs(tableName, rows, columns,
1113 HConstants.LATEST_TIMESTAMP,
1114 attributes);
1115 }
1116
1117 @Override
1118 public List<TRowResult> getRowsTs(ByteBuffer tableName,
1119 List<ByteBuffer> rows,
1120 long timestamp,
1121 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1122 return getRowsWithColumnsTs(tableName, rows, null,
1123 timestamp, attributes);
1124 }
1125
1126 @Override
1127 public List<TRowResult> getRowsWithColumnsTs(ByteBuffer tableName,
1128 List<ByteBuffer> rows,
1129 List<ByteBuffer> columns, long timestamp,
1130 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1131
1132 Table table= null;
1133 try {
1134 List<Get> gets = new ArrayList<Get>(rows.size());
1135 table = getTable(tableName);
1136 if (metrics != null) {
1137 metrics.incNumRowKeysInBatchGet(rows.size());
1138 }
1139 for (ByteBuffer row : rows) {
1140 Get get = new Get(getBytes(row));
1141 addAttributes(get, attributes);
1142 if (columns != null) {
1143
1144 for(ByteBuffer column : columns) {
1145 byte [][] famAndQf = KeyValue.parseColumn(getBytes(column));
1146 if (famAndQf.length == 1) {
1147 get.addFamily(famAndQf[0]);
1148 } else {
1149 get.addColumn(famAndQf[0], famAndQf[1]);
1150 }
1151 }
1152 }
1153 get.setTimeRange(0, timestamp);
1154 gets.add(get);
1155 }
1156 Result[] result = table.get(gets);
1157 return ThriftUtilities.rowResultFromHBase(result);
1158 } catch (IOException e) {
1159 LOG.warn(e.getMessage(), e);
1160 throw getIOError(e);
1161 } finally{
1162 closeTable(table);
1163 }
1164 }
1165
1166 @Override
1167 public void deleteAll(
1168 ByteBuffer tableName, ByteBuffer row, ByteBuffer column,
1169 Map<ByteBuffer, ByteBuffer> attributes)
1170 throws IOError {
1171 deleteAllTs(tableName, row, column, HConstants.LATEST_TIMESTAMP,
1172 attributes);
1173 }
1174
1175 @Override
1176 public void deleteAllTs(ByteBuffer tableName,
1177 ByteBuffer row,
1178 ByteBuffer column,
1179 long timestamp, Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1180 Table table = null;
1181 try {
1182 table = getTable(tableName);
1183 Delete delete = new Delete(getBytes(row));
1184 addAttributes(delete, attributes);
1185 byte [][] famAndQf = KeyValue.parseColumn(getBytes(column));
1186 if (famAndQf.length == 1) {
1187 delete.deleteFamily(famAndQf[0], timestamp);
1188 } else {
1189 delete.deleteColumns(famAndQf[0], famAndQf[1], timestamp);
1190 }
1191 table.delete(delete);
1192
1193 } catch (IOException e) {
1194 LOG.warn(e.getMessage(), e);
1195 throw getIOError(e);
1196 } finally {
1197 closeTable(table);
1198 }
1199 }
1200
1201 @Override
1202 public void deleteAllRow(
1203 ByteBuffer tableName, ByteBuffer row,
1204 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1205 deleteAllRowTs(tableName, row, HConstants.LATEST_TIMESTAMP, attributes);
1206 }
1207
1208 @Override
1209 public void deleteAllRowTs(
1210 ByteBuffer tableName, ByteBuffer row, long timestamp,
1211 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1212 Table table = null;
1213 try {
1214 table = getTable(tableName);
1215 Delete delete = new Delete(getBytes(row), timestamp);
1216 addAttributes(delete, attributes);
1217 table.delete(delete);
1218 } catch (IOException e) {
1219 LOG.warn(e.getMessage(), e);
1220 throw getIOError(e);
1221 } finally {
1222 closeTable(table);
1223 }
1224 }
1225
1226 @Override
1227 public void createTable(ByteBuffer in_tableName,
1228 List<ColumnDescriptor> columnFamilies) throws IOError,
1229 IllegalArgument, AlreadyExists {
1230 TableName tableName = getTableName(in_tableName);
1231 try {
1232 if (getAdmin().tableExists(tableName)) {
1233 throw new AlreadyExists("table name already in use");
1234 }
1235 HTableDescriptor desc = new HTableDescriptor(tableName);
1236 for (ColumnDescriptor col : columnFamilies) {
1237 HColumnDescriptor colDesc = ThriftUtilities.colDescFromThrift(col);
1238 desc.addFamily(colDesc);
1239 }
1240 getAdmin().createTable(desc);
1241 } catch (IOException e) {
1242 LOG.warn(e.getMessage(), e);
1243 throw getIOError(e);
1244 } catch (IllegalArgumentException e) {
1245 LOG.warn(e.getMessage(), e);
1246 throw new IllegalArgument(Throwables.getStackTraceAsString(e));
1247 }
1248 }
1249
1250 private static TableName getTableName(ByteBuffer buffer) {
1251 return TableName.valueOf(getBytes(buffer));
1252 }
1253
1254 @Override
1255 public void deleteTable(ByteBuffer in_tableName) throws IOError {
1256 TableName tableName = getTableName(in_tableName);
1257 if (LOG.isDebugEnabled()) {
1258 LOG.debug("deleteTable: table=" + tableName);
1259 }
1260 try {
1261 if (!getAdmin().tableExists(tableName)) {
1262 throw new IOException("table does not exist");
1263 }
1264 getAdmin().deleteTable(tableName);
1265 } catch (IOException e) {
1266 LOG.warn(e.getMessage(), e);
1267 throw getIOError(e);
1268 }
1269 }
1270
1271 @Override
1272 public void mutateRow(ByteBuffer tableName, ByteBuffer row,
1273 List<Mutation> mutations, Map<ByteBuffer, ByteBuffer> attributes)
1274 throws IOError, IllegalArgument {
1275 mutateRowTs(tableName, row, mutations, HConstants.LATEST_TIMESTAMP,
1276 attributes);
1277 }
1278
1279 @Override
1280 public void mutateRowTs(ByteBuffer tableName, ByteBuffer row,
1281 List<Mutation> mutations, long timestamp,
1282 Map<ByteBuffer, ByteBuffer> attributes)
1283 throws IOError, IllegalArgument {
1284 Table table = null;
1285 try {
1286 table = getTable(tableName);
1287 Put put = new Put(getBytes(row), timestamp);
1288 addAttributes(put, attributes);
1289
1290 Delete delete = new Delete(getBytes(row));
1291 addAttributes(delete, attributes);
1292 if (metrics != null) {
1293 metrics.incNumRowKeysInBatchMutate(mutations.size());
1294 }
1295
1296
1297 for (Mutation m : mutations) {
1298 byte[][] famAndQf = KeyValue.parseColumn(getBytes(m.column));
1299 if (m.isDelete) {
1300 if (famAndQf.length == 1) {
1301 delete.deleteFamily(famAndQf[0], timestamp);
1302 } else {
1303 delete.deleteColumns(famAndQf[0], famAndQf[1], timestamp);
1304 }
1305 delete.setDurability(m.writeToWAL ? Durability.SYNC_WAL
1306 : Durability.SKIP_WAL);
1307 } else {
1308 if(famAndQf.length == 1) {
1309 LOG.warn("No column qualifier specified. Delete is the only mutation supported "
1310 + "over the whole column family.");
1311 } else {
1312 put.addImmutable(famAndQf[0], famAndQf[1],
1313 m.value != null ? getBytes(m.value)
1314 : HConstants.EMPTY_BYTE_ARRAY);
1315 }
1316 put.setDurability(m.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1317 }
1318 }
1319 if (!delete.isEmpty())
1320 table.delete(delete);
1321 if (!put.isEmpty())
1322 table.put(put);
1323 } catch (IOException e) {
1324 LOG.warn(e.getMessage(), e);
1325 throw getIOError(e);
1326 } catch (IllegalArgumentException e) {
1327 LOG.warn(e.getMessage(), e);
1328 throw new IllegalArgument(Throwables.getStackTraceAsString(e));
1329 } finally{
1330 closeTable(table);
1331 }
1332 }
1333
1334 @Override
1335 public void mutateRows(ByteBuffer tableName, List<BatchMutation> rowBatches,
1336 Map<ByteBuffer, ByteBuffer> attributes)
1337 throws IOError, IllegalArgument, TException {
1338 mutateRowsTs(tableName, rowBatches, HConstants.LATEST_TIMESTAMP, attributes);
1339 }
1340
1341 @Override
1342 public void mutateRowsTs(
1343 ByteBuffer tableName, List<BatchMutation> rowBatches, long timestamp,
1344 Map<ByteBuffer, ByteBuffer> attributes)
1345 throws IOError, IllegalArgument, TException {
1346 List<Put> puts = new ArrayList<Put>();
1347 List<Delete> deletes = new ArrayList<Delete>();
1348
1349 for (BatchMutation batch : rowBatches) {
1350 byte[] row = getBytes(batch.row);
1351 List<Mutation> mutations = batch.mutations;
1352 Delete delete = new Delete(row);
1353 addAttributes(delete, attributes);
1354 Put put = new Put(row, timestamp);
1355 addAttributes(put, attributes);
1356 for (Mutation m : mutations) {
1357 byte[][] famAndQf = KeyValue.parseColumn(getBytes(m.column));
1358 if (m.isDelete) {
1359
1360 if (famAndQf.length == 1) {
1361 delete.deleteFamily(famAndQf[0], timestamp);
1362 } else {
1363 delete.deleteColumns(famAndQf[0], famAndQf[1], timestamp);
1364 }
1365 delete.setDurability(m.writeToWAL ? Durability.SYNC_WAL
1366 : Durability.SKIP_WAL);
1367 } else {
1368 if (famAndQf.length == 1) {
1369 LOG.warn("No column qualifier specified. Delete is the only mutation supported "
1370 + "over the whole column family.");
1371 }
1372 if (famAndQf.length == 2) {
1373 put.addImmutable(famAndQf[0], famAndQf[1],
1374 m.value != null ? getBytes(m.value)
1375 : HConstants.EMPTY_BYTE_ARRAY);
1376 } else {
1377 throw new IllegalArgumentException("Invalid famAndQf provided.");
1378 }
1379 put.setDurability(m.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1380 }
1381 }
1382 if (!delete.isEmpty())
1383 deletes.add(delete);
1384 if (!put.isEmpty())
1385 puts.add(put);
1386 }
1387
1388 Table table = null;
1389 try {
1390 table = getTable(tableName);
1391 if (!puts.isEmpty())
1392 table.put(puts);
1393 if (!deletes.isEmpty())
1394 table.delete(deletes);
1395
1396 } catch (IOException e) {
1397 LOG.warn(e.getMessage(), e);
1398 throw getIOError(e);
1399 } catch (IllegalArgumentException e) {
1400 LOG.warn(e.getMessage(), e);
1401 throw new IllegalArgument(Throwables.getStackTraceAsString(e));
1402 } finally{
1403 closeTable(table);
1404 }
1405 }
1406
1407 @Override
1408 public long atomicIncrement(
1409 ByteBuffer tableName, ByteBuffer row, ByteBuffer column, long amount)
1410 throws IOError, IllegalArgument, TException {
1411 byte [][] famAndQf = KeyValue.parseColumn(getBytes(column));
1412 if(famAndQf.length == 1) {
1413 return atomicIncrement(tableName, row, famAndQf[0], HConstants.EMPTY_BYTE_ARRAY, amount);
1414 }
1415 return atomicIncrement(tableName, row, famAndQf[0], famAndQf[1], amount);
1416 }
1417
1418 protected long atomicIncrement(ByteBuffer tableName, ByteBuffer row,
1419 byte [] family, byte [] qualifier, long amount)
1420 throws IOError, IllegalArgument, TException {
1421 Table table = null;
1422 try {
1423 table = getTable(tableName);
1424 return table.incrementColumnValue(
1425 getBytes(row), family, qualifier, amount);
1426 } catch (IOException e) {
1427 LOG.warn(e.getMessage(), e);
1428 throw getIOError(e);
1429 } finally {
1430 closeTable(table);
1431 }
1432 }
1433
1434 @Override
1435 public void scannerClose(int id) throws IOError, IllegalArgument {
1436 LOG.debug("scannerClose: id=" + id);
1437 ResultScannerWrapper resultScannerWrapper = getScanner(id);
1438 if (resultScannerWrapper == null) {
1439 String message = "scanner ID is invalid";
1440 LOG.warn(message);
1441 throw new IllegalArgument("scanner ID is invalid");
1442 }
1443 resultScannerWrapper.getScanner().close();
1444 removeScanner(id);
1445 }
1446
1447 @Override
1448 public List<TRowResult> scannerGetList(int id,int nbRows)
1449 throws IllegalArgument, IOError {
1450 LOG.debug("scannerGetList: id=" + id);
1451 ResultScannerWrapper resultScannerWrapper = getScanner(id);
1452 if (null == resultScannerWrapper) {
1453 String message = "scanner ID is invalid";
1454 LOG.warn(message);
1455 throw new IllegalArgument("scanner ID is invalid");
1456 }
1457
1458 Result [] results = null;
1459 try {
1460 results = resultScannerWrapper.getScanner().next(nbRows);
1461 if (null == results) {
1462 return new ArrayList<TRowResult>();
1463 }
1464 } catch (IOException e) {
1465 LOG.warn(e.getMessage(), e);
1466 throw getIOError(e);
1467 }
1468 return ThriftUtilities.rowResultFromHBase(results, resultScannerWrapper.isColumnSorted());
1469 }
1470
1471 @Override
1472 public List<TRowResult> scannerGet(int id) throws IllegalArgument, IOError {
1473 return scannerGetList(id,1);
1474 }
1475
1476 @Override
1477 public int scannerOpenWithScan(ByteBuffer tableName, TScan tScan,
1478 Map<ByteBuffer, ByteBuffer> attributes)
1479 throws IOError {
1480
1481 Table table = null;
1482 try {
1483 table = getTable(tableName);
1484 Scan scan = new Scan();
1485 addAttributes(scan, attributes);
1486 if (tScan.isSetStartRow()) {
1487 scan.setStartRow(tScan.getStartRow());
1488 }
1489 if (tScan.isSetStopRow()) {
1490 scan.setStopRow(tScan.getStopRow());
1491 }
1492 if (tScan.isSetTimestamp()) {
1493 scan.setTimeRange(0, tScan.getTimestamp());
1494 }
1495 if (tScan.isSetCaching()) {
1496 scan.setCaching(tScan.getCaching());
1497 }
1498 if (tScan.isSetBatchSize()) {
1499 scan.setBatch(tScan.getBatchSize());
1500 }
1501 if (tScan.isSetColumns() && tScan.getColumns().size() != 0) {
1502 for(ByteBuffer column : tScan.getColumns()) {
1503 byte [][] famQf = KeyValue.parseColumn(getBytes(column));
1504 if(famQf.length == 1) {
1505 scan.addFamily(famQf[0]);
1506 } else {
1507 scan.addColumn(famQf[0], famQf[1]);
1508 }
1509 }
1510 }
1511 if (tScan.isSetFilterString()) {
1512 ParseFilter parseFilter = new ParseFilter();
1513 scan.setFilter(
1514 parseFilter.parseFilterString(tScan.getFilterString()));
1515 }
1516 if (tScan.isSetReversed()) {
1517 scan.setReversed(tScan.isReversed());
1518 }
1519 if (tScan.isSetCacheBlocks()) {
1520 scan.setCacheBlocks(tScan.isCacheBlocks());
1521 }
1522 return addScanner(table.getScanner(scan), tScan.sortColumns);
1523 } catch (IOException e) {
1524 LOG.warn(e.getMessage(), e);
1525 throw getIOError(e);
1526 } finally{
1527 closeTable(table);
1528 }
1529 }
1530
1531 @Override
1532 public int scannerOpen(ByteBuffer tableName, ByteBuffer startRow,
1533 List<ByteBuffer> columns,
1534 Map<ByteBuffer, ByteBuffer> attributes) throws IOError {
1535
1536 Table table = null;
1537 try {
1538 table = getTable(tableName);
1539 Scan scan = new Scan(getBytes(startRow));
1540 addAttributes(scan, attributes);
1541 if(columns != null && columns.size() != 0) {
1542 for(ByteBuffer column : columns) {
1543 byte [][] famQf = KeyValue.parseColumn(getBytes(column));
1544 if(famQf.length == 1) {
1545 scan.addFamily(famQf[0]);
1546 } else {
1547 scan.addColumn(famQf[0], famQf[1]);
1548 }
1549 }
1550 }
1551 return addScanner(table.getScanner(scan), false);
1552 } catch (IOException e) {
1553 LOG.warn(e.getMessage(), e);
1554 throw getIOError(e);
1555 } finally{
1556 closeTable(table);
1557 }
1558 }
1559
1560 @Override
1561 public int scannerOpenWithStop(ByteBuffer tableName, ByteBuffer startRow,
1562 ByteBuffer stopRow, List<ByteBuffer> columns,
1563 Map<ByteBuffer, ByteBuffer> attributes)
1564 throws IOError, TException {
1565
1566 Table table = null;
1567 try {
1568 table = getTable(tableName);
1569 Scan scan = new Scan(getBytes(startRow), getBytes(stopRow));
1570 addAttributes(scan, attributes);
1571 if(columns != null && columns.size() != 0) {
1572 for(ByteBuffer column : columns) {
1573 byte [][] famQf = KeyValue.parseColumn(getBytes(column));
1574 if(famQf.length == 1) {
1575 scan.addFamily(famQf[0]);
1576 } else {
1577 scan.addColumn(famQf[0], famQf[1]);
1578 }
1579 }
1580 }
1581 return addScanner(table.getScanner(scan), false);
1582 } catch (IOException e) {
1583 LOG.warn(e.getMessage(), e);
1584 throw getIOError(e);
1585 } finally{
1586 closeTable(table);
1587 }
1588 }
1589
1590 @Override
1591 public int scannerOpenWithPrefix(ByteBuffer tableName,
1592 ByteBuffer startAndPrefix,
1593 List<ByteBuffer> columns,
1594 Map<ByteBuffer, ByteBuffer> attributes)
1595 throws IOError, TException {
1596
1597 Table table = null;
1598 try {
1599 table = getTable(tableName);
1600 Scan scan = new Scan(getBytes(startAndPrefix));
1601 addAttributes(scan, attributes);
1602 Filter f = new WhileMatchFilter(
1603 new PrefixFilter(getBytes(startAndPrefix)));
1604 scan.setFilter(f);
1605 if (columns != null && columns.size() != 0) {
1606 for(ByteBuffer column : columns) {
1607 byte [][] famQf = KeyValue.parseColumn(getBytes(column));
1608 if(famQf.length == 1) {
1609 scan.addFamily(famQf[0]);
1610 } else {
1611 scan.addColumn(famQf[0], famQf[1]);
1612 }
1613 }
1614 }
1615 return addScanner(table.getScanner(scan), false);
1616 } catch (IOException e) {
1617 LOG.warn(e.getMessage(), e);
1618 throw getIOError(e);
1619 } finally{
1620 closeTable(table);
1621 }
1622 }
1623
1624 @Override
1625 public int scannerOpenTs(ByteBuffer tableName, ByteBuffer startRow,
1626 List<ByteBuffer> columns, long timestamp,
1627 Map<ByteBuffer, ByteBuffer> attributes) throws IOError, TException {
1628
1629 Table table = null;
1630 try {
1631 table = getTable(tableName);
1632 Scan scan = new Scan(getBytes(startRow));
1633 addAttributes(scan, attributes);
1634 scan.setTimeRange(0, timestamp);
1635 if (columns != null && columns.size() != 0) {
1636 for (ByteBuffer column : columns) {
1637 byte [][] famQf = KeyValue.parseColumn(getBytes(column));
1638 if(famQf.length == 1) {
1639 scan.addFamily(famQf[0]);
1640 } else {
1641 scan.addColumn(famQf[0], famQf[1]);
1642 }
1643 }
1644 }
1645 return addScanner(table.getScanner(scan), false);
1646 } catch (IOException e) {
1647 LOG.warn(e.getMessage(), e);
1648 throw getIOError(e);
1649 } finally{
1650 closeTable(table);
1651 }
1652 }
1653
1654 @Override
1655 public int scannerOpenWithStopTs(ByteBuffer tableName, ByteBuffer startRow,
1656 ByteBuffer stopRow, List<ByteBuffer> columns, long timestamp,
1657 Map<ByteBuffer, ByteBuffer> attributes)
1658 throws IOError, TException {
1659
1660 Table table = null;
1661 try {
1662 table = getTable(tableName);
1663 Scan scan = new Scan(getBytes(startRow), getBytes(stopRow));
1664 addAttributes(scan, attributes);
1665 scan.setTimeRange(0, timestamp);
1666 if (columns != null && columns.size() != 0) {
1667 for (ByteBuffer column : columns) {
1668 byte [][] famQf = KeyValue.parseColumn(getBytes(column));
1669 if(famQf.length == 1) {
1670 scan.addFamily(famQf[0]);
1671 } else {
1672 scan.addColumn(famQf[0], famQf[1]);
1673 }
1674 }
1675 }
1676 scan.setTimeRange(0, timestamp);
1677 return addScanner(table.getScanner(scan), false);
1678 } catch (IOException e) {
1679 LOG.warn(e.getMessage(), e);
1680 throw getIOError(e);
1681 } finally{
1682 closeTable(table);
1683 }
1684 }
1685
1686 @Override
1687 public Map<ByteBuffer, ColumnDescriptor> getColumnDescriptors(
1688 ByteBuffer tableName) throws IOError, TException {
1689
1690 Table table = null;
1691 try {
1692 TreeMap<ByteBuffer, ColumnDescriptor> columns =
1693 new TreeMap<ByteBuffer, ColumnDescriptor>();
1694
1695 table = getTable(tableName);
1696 HTableDescriptor desc = table.getTableDescriptor();
1697
1698 for (HColumnDescriptor e : desc.getFamilies()) {
1699 ColumnDescriptor col = ThriftUtilities.colDescFromHbase(e);
1700 columns.put(col.name, col);
1701 }
1702 return columns;
1703 } catch (IOException e) {
1704 LOG.warn(e.getMessage(), e);
1705 throw getIOError(e);
1706 } finally {
1707 closeTable(table);
1708 }
1709 }
1710
1711 @Deprecated
1712 @Override
1713 public List<TCell> getRowOrBefore(ByteBuffer tableName, ByteBuffer row,
1714 ByteBuffer family) throws IOError {
1715 try {
1716 Result result = getRowOrBefore(getBytes(tableName), getBytes(row), getBytes(family));
1717 return ThriftUtilities.cellFromHBase(result.rawCells());
1718 } catch (IOException e) {
1719 LOG.warn(e.getMessage(), e);
1720 throw getIOError(e);
1721 }
1722 }
1723
1724 @Override
1725 public TRegionInfo getRegionInfo(ByteBuffer searchRow) throws IOError {
1726 try {
1727 byte[] row = getBytes(searchRow);
1728 Result startRowResult =
1729 getRowOrBefore(TableName.META_TABLE_NAME.getName(), row, HConstants.CATALOG_FAMILY);
1730
1731 if (startRowResult == null) {
1732 throw new IOException("Cannot find row in "+ TableName.META_TABLE_NAME+", row="
1733 + Bytes.toStringBinary(row));
1734 }
1735
1736
1737 HRegionInfo regionInfo = HRegionInfo.getHRegionInfo(startRowResult);
1738 if (regionInfo == null) {
1739 throw new IOException("HRegionInfo REGIONINFO was null or " +
1740 " empty in Meta for row="
1741 + Bytes.toStringBinary(row));
1742 }
1743 TRegionInfo region = new TRegionInfo();
1744 region.setStartKey(regionInfo.getStartKey());
1745 region.setEndKey(regionInfo.getEndKey());
1746 region.id = regionInfo.getRegionId();
1747 region.setName(regionInfo.getRegionName());
1748 region.version = regionInfo.getVersion();
1749
1750
1751 ServerName serverName = HRegionInfo.getServerName(startRowResult);
1752 if (serverName != null) {
1753 region.setServerName(Bytes.toBytes(serverName.getHostname()));
1754 region.port = serverName.getPort();
1755 }
1756 return region;
1757 } catch (IOException e) {
1758 LOG.warn(e.getMessage(), e);
1759 throw getIOError(e);
1760 }
1761 }
1762
1763 private void closeTable(Table table) throws IOError
1764 {
1765 try{
1766 if(table != null){
1767 table.close();
1768 }
1769 } catch (IOException e){
1770 LOG.error(e.getMessage(), e);
1771 throw new IOError(Throwables.getStackTraceAsString(e));
1772 }
1773 }
1774
1775 private Result getRowOrBefore(byte[] tableName, byte[] row, byte[] family) throws IOException {
1776 Scan scan = new Scan(row);
1777 scan.setReversed(true);
1778 scan.addFamily(family);
1779 scan.setStartRow(row);
1780 Table table = getTable(tableName);
1781 try (ResultScanner scanner = table.getScanner(scan)) {
1782 return scanner.next();
1783 } finally{
1784 if(table != null){
1785 table.close();
1786 }
1787 }
1788 }
1789
1790 private void initMetrics(ThriftMetrics metrics) {
1791 this.metrics = metrics;
1792 }
1793
1794 @Override
1795 public void increment(TIncrement tincrement) throws IOError, TException {
1796
1797 if (tincrement.getRow().length == 0 || tincrement.getTable().length == 0) {
1798 throw new TException("Must supply a table and a row key; can't increment");
1799 }
1800
1801 if (conf.getBoolean(COALESCE_INC_KEY, false)) {
1802 this.coalescer.queueIncrement(tincrement);
1803 return;
1804 }
1805
1806 Table table = null;
1807 try {
1808 table = getTable(tincrement.getTable());
1809 Increment inc = ThriftUtilities.incrementFromThrift(tincrement);
1810 table.increment(inc);
1811 } catch (IOException e) {
1812 LOG.warn(e.getMessage(), e);
1813 throw getIOError(e);
1814 } finally{
1815 closeTable(table);
1816 }
1817 }
1818
1819 @Override
1820 public void incrementRows(List<TIncrement> tincrements) throws IOError, TException {
1821 if (conf.getBoolean(COALESCE_INC_KEY, false)) {
1822 this.coalescer.queueIncrements(tincrements);
1823 return;
1824 }
1825 for (TIncrement tinc : tincrements) {
1826 increment(tinc);
1827 }
1828 }
1829
1830 @Override
1831 public List<TCell> append(TAppend tappend) throws IOError, TException {
1832 if (tappend.getRow().length == 0 || tappend.getTable().length == 0) {
1833 throw new TException("Must supply a table and a row key; can't append");
1834 }
1835
1836 Table table = null;
1837 try {
1838 table = getTable(tappend.getTable());
1839 Append append = ThriftUtilities.appendFromThrift(tappend);
1840 Result result = table.append(append);
1841 return ThriftUtilities.cellFromHBase(result.rawCells());
1842 } catch (IOException e) {
1843 LOG.warn(e.getMessage(), e);
1844 throw getIOError(e);
1845 } finally{
1846 closeTable(table);
1847 }
1848 }
1849
1850 @Override
1851 public boolean checkAndPut(ByteBuffer tableName, ByteBuffer row, ByteBuffer column,
1852 ByteBuffer value, Mutation mput, Map<ByteBuffer, ByteBuffer> attributes) throws IOError,
1853 IllegalArgument, TException {
1854 Put put;
1855 try {
1856 put = new Put(getBytes(row), HConstants.LATEST_TIMESTAMP);
1857 addAttributes(put, attributes);
1858
1859 byte[][] famAndQf = KeyValue.parseColumn(getBytes(mput.column));
1860
1861 put.addImmutable(famAndQf[0], famAndQf[1], mput.value != null ? getBytes(mput.value)
1862 : HConstants.EMPTY_BYTE_ARRAY);
1863
1864 put.setDurability(mput.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1865 } catch (IllegalArgumentException e) {
1866 LOG.warn(e.getMessage(), e);
1867 throw new IllegalArgument(Throwables.getStackTraceAsString(e));
1868 }
1869
1870 Table table = null;
1871 try {
1872 table = getTable(tableName);
1873 byte[][] famAndQf = KeyValue.parseColumn(getBytes(column));
1874 return table.checkAndPut(getBytes(row), famAndQf[0], famAndQf[1],
1875 value != null ? getBytes(value) : HConstants.EMPTY_BYTE_ARRAY, put);
1876 } catch (IOException e) {
1877 LOG.warn(e.getMessage(), e);
1878 throw getIOError(e);
1879 } catch (IllegalArgumentException e) {
1880 LOG.warn(e.getMessage(), e);
1881 throw new IllegalArgument(Throwables.getStackTraceAsString(e));
1882 } finally {
1883 closeTable(table);
1884 }
1885 }
1886 }
1887
1888
1889 private static IOError getIOError(Throwable throwable) {
1890 IOError error = new IOErrorWithCause(throwable);
1891 error.setMessage(Throwables.getStackTraceAsString(throwable));
1892 return error;
1893 }
1894
1895
1896
1897
1898 private static void addAttributes(OperationWithAttributes op,
1899 Map<ByteBuffer, ByteBuffer> attributes) {
1900 if (attributes == null || attributes.size() == 0) {
1901 return;
1902 }
1903 for (Map.Entry<ByteBuffer, ByteBuffer> entry : attributes.entrySet()) {
1904 String name = Bytes.toStringBinary(getBytes(entry.getKey()));
1905 byte[] value = getBytes(entry.getValue());
1906 op.setAttribute(name, value);
1907 }
1908 }
1909
1910 public static void registerFilters(Configuration conf) {
1911 String[] filters = conf.getStrings("hbase.thrift.filters");
1912 if(filters != null) {
1913 for(String filterClass: filters) {
1914 String[] filterPart = filterClass.split(":");
1915 if(filterPart.length != 2) {
1916 LOG.warn("Invalid filter specification " + filterClass + " - skipping");
1917 } else {
1918 ParseFilter.registerFilter(filterPart[0], filterPart[1]);
1919 }
1920 }
1921 }
1922 }
1923
1924 public static class IOErrorWithCause extends IOError {
1925 private static final long serialVersionUID = 3545290006843153883L;
1926
1927 private Throwable cause;
1928 public IOErrorWithCause(Throwable cause) {
1929 this.cause = cause;
1930 }
1931
1932 @Override
1933 public synchronized Throwable getCause() {
1934 return cause;
1935 }
1936
1937 @Override
1938 public boolean equals(Object other) {
1939 if (super.equals(other) &&
1940 other instanceof IOErrorWithCause) {
1941 Throwable otherCause = ((IOErrorWithCause) other).getCause();
1942 if (this.getCause() != null) {
1943 return otherCause != null && this.getCause().equals(otherCause);
1944 } else {
1945 return otherCause == null;
1946 }
1947 }
1948 return false;
1949 }
1950
1951 @Override
1952 public int hashCode() {
1953 int result = super.hashCode();
1954 result = 31 * result + (cause != null ? cause.hashCode() : 0);
1955 return result;
1956 }
1957 }
1958 }