View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
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   * This class is a glue object that connects Thrift RPC calls to the HBase client API primarily
77   * defined in the HTableInterface.
78   */
79  @InterfaceAudience.Private
80  public class ThriftHBaseServiceHandler implements THBaseService.Iface {
81  
82    // TODO: Size of pool configuraple
83    private static final Log LOG = LogFactory.getLog(ThriftHBaseServiceHandler.class);
84  
85    // nextScannerId and scannerMap are used to manage scanner state
86    // TODO: Cleanup thread for Scanners, Scanner id wrap
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    * Assigns a unique ID to the scanner and adds the mapping to an internal HashMap.
210    * @param scanner to add
211    * @return Id for this Scanner
212    */
213   private int addScanner(ResultScanner scanner) {
214     int id = nextScannerId.getAndIncrement();
215     scannerMap.put(id, scanner);
216     return id;
217   }
218 
219   /**
220    * Returns the Scanner associated with the specified Id.
221    * @param id of the Scanner to get
222    * @return a Scanner, or null if the Id is invalid
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    * Removes the scanner associated with the specified ID from the internal HashMap.
234    * @param id of the Scanner to remove
235    * @return the removed Scanner, or <code>null</code> if the Id is invalid
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 }