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