1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.thrift2;
20
21 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.appendFromThrift;
22 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.compareOpFromThrift;
23 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.deleteFromThrift;
24 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.deletesFromThrift;
25 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.getFromThrift;
26 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.getsFromThrift;
27 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.incrementFromThrift;
28 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.putFromThrift;
29 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.putsFromThrift;
30 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.resultFromHBase;
31 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.resultsFromHBase;
32 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.rowMutationsFromThrift;
33 import static org.apache.hadoop.hbase.thrift2.ThriftUtilities.scanFromThrift;
34 import static org.apache.thrift.TBaseHelper.byteBufferToByteArray;
35
36 import java.io.IOException;
37 import java.lang.reflect.InvocationHandler;
38 import java.lang.reflect.InvocationTargetException;
39 import java.lang.reflect.Method;
40 import java.lang.reflect.Proxy;
41 import java.nio.ByteBuffer;
42 import java.util.ArrayList;
43 import java.util.Collections;
44 import java.util.List;
45 import java.util.Map;
46 import java.util.concurrent.ConcurrentHashMap;
47 import java.util.concurrent.atomic.AtomicInteger;
48
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.DoNotRetryIOException;
53 import org.apache.hadoop.hbase.HRegionLocation;
54 import org.apache.hadoop.hbase.classification.InterfaceAudience;
55 import org.apache.hadoop.hbase.client.RegionLocator;
56 import org.apache.hadoop.hbase.client.ResultScanner;
57 import org.apache.hadoop.hbase.client.Table;
58 import org.apache.hadoop.hbase.security.UserProvider;
59 import org.apache.hadoop.hbase.thrift.ThriftMetrics;
60 import org.apache.hadoop.hbase.thrift2.generated.TAppend;
61 import org.apache.hadoop.hbase.thrift2.generated.TCompareOp;
62 import org.apache.hadoop.hbase.thrift2.generated.TDelete;
63 import org.apache.hadoop.hbase.thrift2.generated.TGet;
64 import org.apache.hadoop.hbase.thrift2.generated.THBaseService;
65 import org.apache.hadoop.hbase.thrift2.generated.THRegionLocation;
66 import org.apache.hadoop.hbase.thrift2.generated.TIOError;
67 import org.apache.hadoop.hbase.thrift2.generated.TIllegalArgument;
68 import org.apache.hadoop.hbase.thrift2.generated.TIncrement;
69 import org.apache.hadoop.hbase.thrift2.generated.TPut;
70 import org.apache.hadoop.hbase.thrift2.generated.TResult;
71 import org.apache.hadoop.hbase.thrift2.generated.TRowMutations;
72 import org.apache.hadoop.hbase.thrift2.generated.TScan;
73 import org.apache.hadoop.hbase.util.Bytes;
74 import org.apache.hadoop.hbase.util.ConnectionCache;
75 import org.apache.thrift.TException;
76
77
78
79
80
81 @InterfaceAudience.Private
82 public class ThriftHBaseServiceHandler implements THBaseService.Iface {
83
84
85 private static final Log LOG = LogFactory.getLog(ThriftHBaseServiceHandler.class);
86
87
88
89 private final AtomicInteger nextScannerId = new AtomicInteger(0);
90 private final Map<Integer, ResultScanner> scannerMap =
91 new ConcurrentHashMap<Integer, ResultScanner>();
92
93 private final ConnectionCache connectionCache;
94
95 static final String CLEANUP_INTERVAL = "hbase.thrift.connection.cleanup-interval";
96 static final String MAX_IDLETIME = "hbase.thrift.connection.max-idletime";
97
98 private static final IOException ioe
99 = new DoNotRetryIOException("Thrift Server is in Read-only mode.");
100 private boolean isReadOnly;
101
102 public static THBaseService.Iface newInstance(
103 THBaseService.Iface handler, ThriftMetrics metrics) {
104 return (THBaseService.Iface) Proxy.newProxyInstance(handler.getClass().getClassLoader(),
105 new Class[] { THBaseService.Iface.class }, new THBaseServiceMetricsProxy(handler, metrics));
106 }
107
108 private static class THBaseServiceMetricsProxy implements InvocationHandler {
109 private final THBaseService.Iface handler;
110 private final ThriftMetrics metrics;
111
112 private THBaseServiceMetricsProxy(THBaseService.Iface handler, ThriftMetrics metrics) {
113 this.handler = handler;
114 this.metrics = metrics;
115 }
116
117 @Override
118 public Object invoke(Object proxy, Method m, Object[] args) throws Throwable {
119 Object result;
120 long start = now();
121 try {
122 result = m.invoke(handler, args);
123 } catch (InvocationTargetException e) {
124 metrics.exception(e.getCause());
125 throw e.getTargetException();
126 } catch (Exception e) {
127 metrics.exception(e);
128 throw new RuntimeException("unexpected invocation exception: " + e.getMessage());
129 } finally {
130 long processTime = now() - start;
131 metrics.incMethodTime(m.getName(), processTime);
132 }
133 return result;
134 }
135 }
136
137 private static class TIOErrorWithCause extends TIOError {
138 private static final long serialVersionUID = -1164984328968862207L;
139
140 private Throwable cause;
141
142 public TIOErrorWithCause(Throwable cause) {
143 super();
144 this.cause = cause;
145 }
146
147 @Override
148 public synchronized Throwable getCause() {
149 return cause;
150 }
151
152 @Override
153 public boolean equals(Object other) {
154 if (super.equals(other) &&
155 other instanceof TIOErrorWithCause) {
156 Throwable otherCause = ((TIOErrorWithCause) other).getCause();
157 if (this.getCause() != null) {
158 return otherCause != null && this.getCause().equals(otherCause);
159 } else {
160 return otherCause == null;
161 }
162 }
163 return false;
164 }
165
166 @Override
167 public int hashCode() {
168 int result = super.hashCode();
169 result = 31 * result + (cause != null ? cause.hashCode() : 0);
170 return result;
171 }
172 }
173
174 private static long now() {
175 return System.nanoTime();
176 }
177
178 ThriftHBaseServiceHandler(final Configuration conf,
179 final UserProvider userProvider) throws IOException {
180 int cleanInterval = conf.getInt(CLEANUP_INTERVAL, 10 * 1000);
181 int maxIdleTime = conf.getInt(MAX_IDLETIME, 10 * 60 * 1000);
182 connectionCache = new ConnectionCache(
183 conf, userProvider, cleanInterval, maxIdleTime);
184 isReadOnly = conf.getBoolean("hbase.thrift.readonly", false);
185 }
186
187 private Table getTable(ByteBuffer tableName) {
188 try {
189 return connectionCache.getTable(Bytes.toString(byteBufferToByteArray(tableName)));
190 } catch (IOException e) {
191 throw new RuntimeException(e);
192 }
193 }
194
195 private RegionLocator getLocator(ByteBuffer tableName) {
196 try {
197 return connectionCache.getRegionLocator(byteBufferToByteArray(tableName));
198 } catch (IOException ie) {
199 throw new RuntimeException(ie);
200 }
201 }
202
203 private void closeTable(Table table) throws TIOError {
204 try {
205 table.close();
206 } catch (IOException e) {
207 throw getTIOError(e);
208 }
209 }
210
211 private TIOError getTIOError(IOException e) {
212 TIOError err = new TIOErrorWithCause(e);
213 err.setMessage(e.getMessage());
214 return err;
215 }
216
217
218
219
220
221
222 private int addScanner(ResultScanner scanner) {
223 int id = nextScannerId.getAndIncrement();
224 scannerMap.put(id, scanner);
225 return id;
226 }
227
228
229
230
231
232
233 private ResultScanner getScanner(int id) {
234 return scannerMap.get(id);
235 }
236
237 void setEffectiveUser(String effectiveUser) {
238 connectionCache.setEffectiveUser(effectiveUser);
239 }
240
241
242
243
244
245
246 protected ResultScanner removeScanner(int id) {
247 return scannerMap.remove(id);
248 }
249
250 @Override
251 public boolean exists(ByteBuffer table, TGet get) throws TIOError, TException {
252 Table htable = getTable(table);
253 try {
254 return htable.exists(getFromThrift(get));
255 } catch (IOException e) {
256 throw getTIOError(e);
257 } finally {
258 closeTable(htable);
259 }
260 }
261
262 @Override
263 public List<Boolean> existsAll(ByteBuffer table, List<TGet> gets) throws TIOError, TException {
264 Table htable = getTable(table);
265 try {
266 boolean[] exists = htable.existsAll(getsFromThrift(gets));
267 List<Boolean> result = new ArrayList<>(exists.length);
268 for (boolean exist : exists) {
269 result.add(exist);
270 }
271 return result;
272 } catch (IOException e) {
273 throw getTIOError(e);
274 } finally {
275 closeTable(htable);
276 }
277 }
278
279 @Override
280 public TResult get(ByteBuffer table, TGet get) throws TIOError, TException {
281 Table htable = getTable(table);
282 try {
283 return resultFromHBase(htable.get(getFromThrift(get)));
284 } catch (IOException e) {
285 throw getTIOError(e);
286 } finally {
287 closeTable(htable);
288 }
289 }
290
291 @Override
292 public List<TResult> getMultiple(ByteBuffer table, List<TGet> gets) throws TIOError, TException {
293 Table htable = getTable(table);
294 try {
295 return resultsFromHBase(htable.get(getsFromThrift(gets)));
296 } catch (IOException e) {
297 throw getTIOError(e);
298 } finally {
299 closeTable(htable);
300 }
301 }
302
303 @Override
304 public void put(ByteBuffer table, TPut put) throws TIOError, TException {
305 checkReadOnlyMode();
306 Table htable = getTable(table);
307 try {
308 htable.put(putFromThrift(put));
309 } catch (IOException e) {
310 throw getTIOError(e);
311 } finally {
312 closeTable(htable);
313 }
314 }
315
316 @Override
317 public boolean checkAndPut(ByteBuffer table, ByteBuffer row, ByteBuffer family,
318 ByteBuffer qualifier, ByteBuffer value, TPut put) throws TIOError, TException {
319 checkReadOnlyMode();
320 Table htable = getTable(table);
321 try {
322 return htable.checkAndPut(byteBufferToByteArray(row), byteBufferToByteArray(family),
323 byteBufferToByteArray(qualifier), (value == null) ? null : byteBufferToByteArray(value),
324 putFromThrift(put));
325 } catch (IOException e) {
326 throw getTIOError(e);
327 } finally {
328 closeTable(htable);
329 }
330 }
331
332 @Override
333 public void putMultiple(ByteBuffer table, List<TPut> puts) throws TIOError, TException {
334 checkReadOnlyMode();
335 Table htable = getTable(table);
336 try {
337 htable.put(putsFromThrift(puts));
338 } catch (IOException e) {
339 throw getTIOError(e);
340 } finally {
341 closeTable(htable);
342 }
343 }
344
345 @Override
346 public void deleteSingle(ByteBuffer table, TDelete deleteSingle) throws TIOError, TException {
347 checkReadOnlyMode();
348 Table htable = getTable(table);
349 try {
350 htable.delete(deleteFromThrift(deleteSingle));
351 } catch (IOException e) {
352 throw getTIOError(e);
353 } finally {
354 closeTable(htable);
355 }
356 }
357
358 @Override
359 public List<TDelete> deleteMultiple(ByteBuffer table, List<TDelete> deletes) throws TIOError,
360 TException {
361 checkReadOnlyMode();
362 Table htable = getTable(table);
363 try {
364 htable.delete(deletesFromThrift(deletes));
365 } catch (IOException e) {
366 throw getTIOError(e);
367 } finally {
368 closeTable(htable);
369 }
370 return Collections.emptyList();
371 }
372
373 @Override
374 public boolean checkAndMutate(ByteBuffer table, ByteBuffer row, ByteBuffer family,
375 ByteBuffer qualifier, TCompareOp compareOp, ByteBuffer value, TRowMutations rowMutations)
376 throws TIOError, TException {
377 checkReadOnlyMode();
378 try (final Table htable = getTable(table)) {
379 return htable.checkAndMutate(byteBufferToByteArray(row), byteBufferToByteArray(family),
380 byteBufferToByteArray(qualifier), compareOpFromThrift(compareOp),
381 byteBufferToByteArray(value), rowMutationsFromThrift(rowMutations));
382 } catch (IOException e) {
383 throw getTIOError(e);
384 }
385 }
386
387 @Override
388 public boolean checkAndDelete(ByteBuffer table, ByteBuffer row, ByteBuffer family,
389 ByteBuffer qualifier, ByteBuffer value, TDelete deleteSingle) throws TIOError, TException {
390 checkReadOnlyMode();
391 Table htable = getTable(table);
392 try {
393 if (value == null) {
394 return htable.checkAndDelete(byteBufferToByteArray(row), byteBufferToByteArray(family),
395 byteBufferToByteArray(qualifier), null, deleteFromThrift(deleteSingle));
396 } else {
397 return htable.checkAndDelete(byteBufferToByteArray(row), byteBufferToByteArray(family),
398 byteBufferToByteArray(qualifier), byteBufferToByteArray(value),
399 deleteFromThrift(deleteSingle));
400 }
401 } catch (IOException e) {
402 throw getTIOError(e);
403 } finally {
404 closeTable(htable);
405 }
406 }
407
408 @Override
409 public TResult increment(ByteBuffer table, TIncrement increment) throws TIOError, TException {
410 checkReadOnlyMode();
411 Table htable = getTable(table);
412 try {
413 return resultFromHBase(htable.increment(incrementFromThrift(increment)));
414 } catch (IOException e) {
415 throw getTIOError(e);
416 } finally {
417 closeTable(htable);
418 }
419 }
420
421 @Override
422 public TResult append(ByteBuffer table, TAppend append) throws TIOError, TException {
423 checkReadOnlyMode();
424 Table htable = getTable(table);
425 try {
426 return resultFromHBase(htable.append(appendFromThrift(append)));
427 } catch (IOException e) {
428 throw getTIOError(e);
429 } finally {
430 closeTable(htable);
431 }
432 }
433
434 @Override
435 public int openScanner(ByteBuffer table, TScan scan) throws TIOError, TException {
436 Table htable = getTable(table);
437 ResultScanner resultScanner = null;
438 try {
439 resultScanner = htable.getScanner(scanFromThrift(scan));
440 } catch (IOException e) {
441 throw getTIOError(e);
442 } finally {
443 closeTable(htable);
444 }
445 return addScanner(resultScanner);
446 }
447
448 @Override
449 public List<TResult> getScannerRows(int scannerId, int numRows) throws TIOError,
450 TIllegalArgument, TException {
451 ResultScanner scanner = getScanner(scannerId);
452 if (scanner == null) {
453 TIllegalArgument ex = new TIllegalArgument();
454 ex.setMessage("Invalid scanner Id");
455 throw ex;
456 }
457 try {
458 connectionCache.updateConnectionAccessTime();
459 return resultsFromHBase(scanner.next(numRows));
460 } catch (IOException e) {
461 throw getTIOError(e);
462 }
463 }
464
465 @Override
466 public List<TResult> getScannerResults(ByteBuffer table, TScan scan, int numRows)
467 throws TIOError, TException {
468 Table htable = getTable(table);
469 List<TResult> results = null;
470 ResultScanner scanner = null;
471 try {
472 scanner = htable.getScanner(scanFromThrift(scan));
473 results = resultsFromHBase(scanner.next(numRows));
474 } catch (IOException e) {
475 throw getTIOError(e);
476 } finally {
477 if (scanner != null) {
478 scanner.close();
479 }
480 closeTable(htable);
481 }
482 return results;
483 }
484
485
486
487 @Override
488 public void closeScanner(int scannerId) throws TIOError, TIllegalArgument, TException {
489 LOG.debug("scannerClose: id=" + scannerId);
490 ResultScanner scanner = getScanner(scannerId);
491 if (scanner == null) {
492 String message = "scanner ID is invalid";
493 LOG.warn(message);
494 TIllegalArgument ex = new TIllegalArgument();
495 ex.setMessage("Invalid scanner Id");
496 throw ex;
497 }
498 scanner.close();
499 removeScanner(scannerId);
500 }
501
502 @Override
503 public void mutateRow(ByteBuffer table, TRowMutations rowMutations) throws TIOError, TException {
504 checkReadOnlyMode();
505 Table htable = getTable(table);
506 try {
507 htable.mutateRow(rowMutationsFromThrift(rowMutations));
508 } catch (IOException e) {
509 throw getTIOError(e);
510 } finally {
511 closeTable(htable);
512 }
513 }
514
515 @Override
516 public List<THRegionLocation> getAllRegionLocations(ByteBuffer table)
517 throws TIOError, TException {
518 RegionLocator locator = null;
519 try {
520 locator = getLocator(table);
521 return ThriftUtilities.regionLocationsFromHBase(locator.getAllRegionLocations());
522
523 } catch (IOException e) {
524 throw getTIOError(e);
525 } finally {
526 if (locator != null) {
527 try {
528 locator.close();
529 } catch (IOException e) {
530 LOG.warn("Couldn't close the locator.", e);
531 }
532 }
533 }
534 }
535
536 @Override
537 public THRegionLocation getRegionLocation(ByteBuffer table, ByteBuffer row, boolean reload)
538 throws TIOError, TException {
539
540 RegionLocator locator = null;
541 try {
542 locator = getLocator(table);
543 byte[] rowBytes = byteBufferToByteArray(row);
544 HRegionLocation hrl = locator.getRegionLocation(rowBytes, reload);
545 return ThriftUtilities.regionLocationFromHBase(hrl);
546
547 } catch (IOException e) {
548 throw getTIOError(e);
549 } finally {
550 if (locator != null) {
551 try {
552 locator.close();
553 } catch (IOException e) {
554 LOG.warn("Couldn't close the locator.", e);
555 }
556 }
557 }
558 }
559
560 private void checkReadOnlyMode() throws TIOError {
561 if (isReadOnly()) {
562 throw getTIOError(ioe);
563 }
564 }
565
566 private boolean isReadOnly() {
567 return isReadOnly;
568 }
569 }