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