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