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  
20  package org.apache.hadoop.hbase.coprocessor;
21  
22  import java.io.IOException;
23  import java.util.ArrayList;
24  import java.util.Collections;
25  import java.util.Comparator;
26  import java.util.HashSet;
27  import java.util.List;
28  import java.util.Map;
29  import java.util.Set;
30  import java.util.SortedSet;
31  import java.util.TreeSet;
32  import java.util.UUID;
33  import java.util.concurrent.ExecutorService;
34  import java.util.concurrent.atomic.AtomicInteger;
35  
36  import com.google.protobuf.Descriptors;
37  import org.apache.commons.logging.Log;
38  import org.apache.commons.logging.LogFactory;
39  import org.apache.hadoop.classification.InterfaceAudience;
40  import org.apache.hadoop.classification.InterfaceStability;
41  import org.apache.hadoop.conf.Configuration;
42  import org.apache.hadoop.fs.Path;
43  import org.apache.hadoop.hbase.Abortable;
44  import org.apache.hadoop.hbase.Coprocessor;
45  import org.apache.hadoop.hbase.CoprocessorEnvironment;
46  import org.apache.hadoop.hbase.DoNotRetryIOException;
47  import org.apache.hadoop.hbase.HBaseInterfaceAudience;
48  import org.apache.hadoop.hbase.HTableDescriptor;
49  import org.apache.hadoop.hbase.TableName;
50  import org.apache.hadoop.hbase.client.Append;
51  import org.apache.hadoop.hbase.client.CoprocessorHConnection;
52  import org.apache.hadoop.hbase.client.Delete;
53  import org.apache.hadoop.hbase.client.Durability;
54  import org.apache.hadoop.hbase.client.Get;
55  import org.apache.hadoop.hbase.client.HConnection;
56  import org.apache.hadoop.hbase.client.HTable;
57  import org.apache.hadoop.hbase.client.HTableInterface;
58  import org.apache.hadoop.hbase.client.Increment;
59  import org.apache.hadoop.hbase.client.Put;
60  import org.apache.hadoop.hbase.client.Result;
61  import org.apache.hadoop.hbase.client.ResultScanner;
62  import org.apache.hadoop.hbase.client.Row;
63  import org.apache.hadoop.hbase.client.RowMutations;
64  import org.apache.hadoop.hbase.client.Scan;
65  import org.apache.hadoop.hbase.client.coprocessor.Batch;
66  import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
67  import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
68  import org.apache.hadoop.hbase.util.Bytes;
69  import org.apache.hadoop.hbase.util.CoprocessorClassLoader;
70  import org.apache.hadoop.hbase.util.SortedCopyOnWriteSet;
71  import org.apache.hadoop.hbase.util.VersionInfo;
72  import org.apache.hadoop.io.MultipleIOException;
73  
74  import com.google.protobuf.Descriptors.ServiceDescriptor;
75  import com.google.protobuf.Message;
76  import com.google.protobuf.Service;
77  import com.google.protobuf.ServiceException;
78  
79  /**
80   * Provides the common setup framework and runtime services for coprocessor
81   * invocation from HBase services.
82   * @param <E> the specific environment extension that a concrete implementation
83   * provides
84   */
85  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC)
86  @InterfaceStability.Evolving
87  public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
88    public static final String REGION_COPROCESSOR_CONF_KEY =
89        "hbase.coprocessor.region.classes";
90    public static final String REGIONSERVER_COPROCESSOR_CONF_KEY =
91        "hbase.coprocessor.regionserver.classes";
92    public static final String USER_REGION_COPROCESSOR_CONF_KEY =
93        "hbase.coprocessor.user.region.classes";
94    public static final String MASTER_COPROCESSOR_CONF_KEY =
95        "hbase.coprocessor.master.classes";
96    public static final String WAL_COPROCESSOR_CONF_KEY =
97      "hbase.coprocessor.wal.classes";
98    public static final String ABORT_ON_ERROR_KEY = "hbase.coprocessor.abortonerror";
99    public static final boolean DEFAULT_ABORT_ON_ERROR = true;
100 
101   private static final Log LOG = LogFactory.getLog(CoprocessorHost.class);
102   protected Abortable abortable;
103   /** Ordered set of loaded coprocessors with lock */
104   protected SortedSet<E> coprocessors =
105       new SortedCopyOnWriteSet<E>(new EnvironmentPriorityComparator());
106   protected Configuration conf;
107   // unique file prefix to use for local copies of jars when classloading
108   protected String pathPrefix;
109   protected AtomicInteger loadSequence = new AtomicInteger();
110 
111   public CoprocessorHost(Abortable abortable) {
112     this.abortable = abortable;
113     this.pathPrefix = UUID.randomUUID().toString();
114   }
115 
116   /**
117    * Not to be confused with the per-object _coprocessors_ (above),
118    * coprocessorNames is static and stores the set of all coprocessors ever
119    * loaded by any thread in this JVM. It is strictly additive: coprocessors are
120    * added to coprocessorNames, by loadInstance() but are never removed, since
121    * the intention is to preserve a history of all loaded coprocessors for
122    * diagnosis in case of server crash (HBASE-4014).
123    */
124   private static Set<String> coprocessorNames =
125       Collections.synchronizedSet(new HashSet<String>());
126   public static Set<String> getLoadedCoprocessors() {
127       return coprocessorNames;
128   }
129 
130   /**
131    * Used to create a parameter to the HServerLoad constructor so that
132    * HServerLoad can provide information about the coprocessors loaded by this
133    * regionserver.
134    * (HBASE-4070: Improve region server metrics to report loaded coprocessors
135    * to master).
136    */
137   public Set<String> getCoprocessors() {
138     Set<String> returnValue = new TreeSet<String>();
139     for(CoprocessorEnvironment e: coprocessors) {
140       returnValue.add(e.getInstance().getClass().getSimpleName());
141     }
142     return returnValue;
143   }
144 
145   /**
146    * Load system coprocessors. Read the class names from configuration.
147    * Called by constructor.
148    */
149   protected void loadSystemCoprocessors(Configuration conf, String confKey) {
150     Class<?> implClass = null;
151 
152     // load default coprocessors from configure file
153     String[] defaultCPClasses = conf.getStrings(confKey);
154     if (defaultCPClasses == null || defaultCPClasses.length == 0)
155       return;
156 
157     int priority = Coprocessor.PRIORITY_SYSTEM;
158     List<E> configured = new ArrayList<E>();
159     for (String className : defaultCPClasses) {
160       className = className.trim();
161       if (findCoprocessor(className) != null) {
162         continue;
163       }
164       ClassLoader cl = this.getClass().getClassLoader();
165       Thread.currentThread().setContextClassLoader(cl);
166       try {
167         implClass = cl.loadClass(className);
168         configured.add(loadInstance(implClass, Coprocessor.PRIORITY_SYSTEM, conf));
169         LOG.info("System coprocessor " + className + " was loaded " +
170             "successfully with priority (" + priority++ + ").");
171       } catch (Throwable t) {
172         // We always abort if system coprocessors cannot be loaded
173         abortServer(className, t);
174       }
175     }
176 
177     // add entire set to the collection for COW efficiency
178     coprocessors.addAll(configured);
179   }
180 
181   /**
182    * Load a coprocessor implementation into the host
183    * @param path path to implementation jar
184    * @param className the main class name
185    * @param priority chaining priority
186    * @param conf configuration for coprocessor
187    * @throws java.io.IOException Exception
188    */
189   public E load(Path path, String className, int priority,
190       Configuration conf) throws IOException {
191     Class<?> implClass = null;
192     LOG.debug("Loading coprocessor class " + className + " with path " +
193         path + " and priority " + priority);
194 
195     ClassLoader cl = null;
196     if (path == null) {
197       try {
198         implClass = getClass().getClassLoader().loadClass(className);
199       } catch (ClassNotFoundException e) {
200         throw new IOException("No jar path specified for " + className);
201       }
202     } else {
203       cl = CoprocessorClassLoader.getClassLoader(
204         path, getClass().getClassLoader(), pathPrefix, conf);
205       try {
206         implClass = cl.loadClass(className);
207       } catch (ClassNotFoundException e) {
208         throw new IOException("Cannot load external coprocessor class " + className, e);
209       }
210     }
211 
212     //load custom code for coprocessor
213     Thread currentThread = Thread.currentThread();
214     ClassLoader hostClassLoader = currentThread.getContextClassLoader();
215     try{
216       // switch temporarily to the thread classloader for custom CP
217       currentThread.setContextClassLoader(cl);
218       E cpInstance = loadInstance(implClass, priority, conf);
219       return cpInstance;
220     } finally {
221       // restore the fresh (host) classloader
222       currentThread.setContextClassLoader(hostClassLoader);
223     }
224   }
225 
226   /**
227    * @param implClass Implementation class
228    * @param priority priority
229    * @param conf configuration
230    * @throws java.io.IOException Exception
231    */
232   public void load(Class<?> implClass, int priority, Configuration conf)
233       throws IOException {
234     E env = loadInstance(implClass, priority, conf);
235     coprocessors.add(env);
236   }
237 
238   /**
239    * @param implClass Implementation class
240    * @param priority priority
241    * @param conf configuration
242    * @throws java.io.IOException Exception
243    */
244   public E loadInstance(Class<?> implClass, int priority, Configuration conf)
245       throws IOException {
246     if (!Coprocessor.class.isAssignableFrom(implClass)) {
247       throw new IOException("Configured class " + implClass.getName() + " must implement "
248           + Coprocessor.class.getName() + " interface ");
249     }
250 
251     // create the instance
252     Coprocessor impl;
253     Object o = null;
254     try {
255       o = implClass.newInstance();
256       impl = (Coprocessor)o;
257     } catch (InstantiationException e) {
258       throw new IOException(e);
259     } catch (IllegalAccessException e) {
260       throw new IOException(e);
261     }
262     // create the environment
263     E env = createEnvironment(implClass, impl, priority, loadSequence.incrementAndGet(), conf);
264     if (env instanceof Environment) {
265       ((Environment)env).startup();
266     }
267     // HBASE-4014: maintain list of loaded coprocessors for later crash analysis
268     // if server (master or regionserver) aborts.
269     coprocessorNames.add(implClass.getName());
270     return env;
271   }
272 
273   /**
274    * Called when a new Coprocessor class is loaded
275    */
276   public abstract E createEnvironment(Class<?> implClass, Coprocessor instance,
277       int priority, int sequence, Configuration conf);
278 
279   public void shutdown(CoprocessorEnvironment e) {
280     if (e instanceof Environment) {
281       if (LOG.isDebugEnabled()) {
282         LOG.debug("Stop coprocessor " + e.getInstance().getClass().getName());
283       }
284       ((Environment)e).shutdown();
285     } else {
286       LOG.warn("Shutdown called on unknown environment: "+
287           e.getClass().getName());
288     }
289   }
290 
291   /**
292    * Find a coprocessor implementation by class name
293    * @param className the class name
294    * @return the coprocessor, or null if not found
295    */
296   public Coprocessor findCoprocessor(String className) {
297     for (E env: coprocessors) {
298       if (env.getInstance().getClass().getName().equals(className) ||
299           env.getInstance().getClass().getSimpleName().equals(className)) {
300         return env.getInstance();
301       }
302     }
303     return null;
304   }
305 
306   /**
307    * Find a coprocessor environment by class name
308    * @param className the class name
309    * @return the coprocessor, or null if not found
310    */
311   public CoprocessorEnvironment findCoprocessorEnvironment(String className) {
312     for (E env: coprocessors) {
313       if (env.getInstance().getClass().getName().equals(className) ||
314           env.getInstance().getClass().getSimpleName().equals(className)) {
315         return env;
316       }
317     }
318     return null;
319   }
320 
321   /**
322    * Retrieves the set of classloaders used to instantiate Coprocessor classes defined in external
323    * jar files.
324    * @return A set of ClassLoader instances
325    */
326   Set<ClassLoader> getExternalClassLoaders() {
327     Set<ClassLoader> externalClassLoaders = new HashSet<ClassLoader>();
328     final ClassLoader systemClassLoader = this.getClass().getClassLoader();
329     for (E env : coprocessors) {
330       ClassLoader cl = env.getInstance().getClass().getClassLoader();
331       if (cl != systemClassLoader ){
332         //do not include system classloader
333         externalClassLoaders.add(cl);
334       }
335     }
336     return externalClassLoaders;
337   }
338 
339   /**
340    * Environment priority comparator.
341    * Coprocessors are chained in sorted order.
342    */
343   static class EnvironmentPriorityComparator
344       implements Comparator<CoprocessorEnvironment> {
345     public int compare(final CoprocessorEnvironment env1,
346         final CoprocessorEnvironment env2) {
347       if (env1.getPriority() < env2.getPriority()) {
348         return -1;
349       } else if (env1.getPriority() > env2.getPriority()) {
350         return 1;
351       }
352       if (env1.getLoadSequence() < env2.getLoadSequence()) {
353         return -1;
354       } else if (env1.getLoadSequence() > env2.getLoadSequence()) {
355         return 1;
356       }
357       return 0;
358     }
359   }
360 
361   /**
362    * Encapsulation of the environment of each coprocessor
363    */
364   public static class Environment implements CoprocessorEnvironment {
365 
366     /**
367      * A wrapper for HTable. Can be used to restrict privilege.
368      *
369      * Currently it just helps to track tables opened by a Coprocessor and
370      * facilitate close of them if it is aborted.
371      *
372      * We also disallow row locking.
373      *
374      * There is nothing now that will stop a coprocessor from using HTable
375      * objects directly instead of this API, but in the future we intend to
376      * analyze coprocessor implementations as they are loaded and reject those
377      * which attempt to use objects and methods outside the Environment
378      * sandbox.
379      */
380     class HTableWrapper implements HTableInterface {
381 
382       private TableName tableName;
383       private HTable table;
384       private HConnection connection;
385 
386       public HTableWrapper(TableName tableName, HConnection connection, ExecutorService pool)
387           throws IOException {
388         this.tableName = tableName;
389         this.table = new HTable(tableName, connection, pool);
390         this.connection = connection;
391         openTables.add(this);
392       }
393 
394       void internalClose() throws IOException {
395         List<IOException> exceptions = new ArrayList<IOException>(2);
396         try {
397         table.close();
398         } catch (IOException e) {
399           exceptions.add(e);
400         }
401         try {
402           // have to self-manage our connection, as per the HTable contract
403           if (this.connection != null) {
404             this.connection.close();
405           }
406         } catch (IOException e) {
407           exceptions.add(e);
408         }
409         if (!exceptions.isEmpty()) {
410           throw MultipleIOException.createIOException(exceptions);
411         }
412       }
413 
414       public Configuration getConfiguration() {
415         return table.getConfiguration();
416       }
417 
418       public void close() throws IOException {
419         try {
420           internalClose();
421         } finally {
422           openTables.remove(this);
423         }
424       }
425 
426       public Result getRowOrBefore(byte[] row, byte[] family)
427           throws IOException {
428         return table.getRowOrBefore(row, family);
429       }
430 
431       public Result get(Get get) throws IOException {
432         return table.get(get);
433       }
434 
435       public boolean exists(Get get) throws IOException {
436         return table.exists(get);
437       }
438 
439       public Boolean[] exists(List<Get> gets) throws IOException{
440         return table.exists(gets);
441       }
442 
443       public void put(Put put) throws IOException {
444         table.put(put);
445       }
446 
447       public void put(List<Put> puts) throws IOException {
448         table.put(puts);
449       }
450 
451       public void delete(Delete delete) throws IOException {
452         table.delete(delete);
453       }
454 
455       public void delete(List<Delete> deletes) throws IOException {
456         table.delete(deletes);
457       }
458 
459       public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier,
460           byte[] value, Put put) throws IOException {
461         return table.checkAndPut(row, family, qualifier, value, put);
462       }
463 
464       public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier,
465           byte[] value, Delete delete) throws IOException {
466         return table.checkAndDelete(row, family, qualifier, value, delete);
467       }
468 
469       public long incrementColumnValue(byte[] row, byte[] family,
470           byte[] qualifier, long amount) throws IOException {
471         return table.incrementColumnValue(row, family, qualifier, amount);
472       }
473 
474       public long incrementColumnValue(byte[] row, byte[] family,
475           byte[] qualifier, long amount, Durability durability)
476           throws IOException {
477         return table.incrementColumnValue(row, family, qualifier, amount,
478             durability);
479       }
480 
481       @Override
482       public Result append(Append append) throws IOException {
483         return table.append(append);
484       }
485 
486       @Override
487       public Result increment(Increment increment) throws IOException {
488         return table.increment(increment);
489       }
490 
491       public void flushCommits() throws IOException {
492         table.flushCommits();
493       }
494 
495       public boolean isAutoFlush() {
496         return table.isAutoFlush();
497       }
498 
499       public ResultScanner getScanner(Scan scan) throws IOException {
500         return table.getScanner(scan);
501       }
502 
503       public ResultScanner getScanner(byte[] family) throws IOException {
504         return table.getScanner(family);
505       }
506 
507       public ResultScanner getScanner(byte[] family, byte[] qualifier)
508           throws IOException {
509         return table.getScanner(family, qualifier);
510       }
511 
512       public HTableDescriptor getTableDescriptor() throws IOException {
513         return table.getTableDescriptor();
514       }
515 
516       @Override
517       public byte[] getTableName() {
518         return tableName.getName();
519       }
520 
521       @Override
522       public TableName getName() {
523         return table.getName();
524       }
525 
526       @Override
527       public void batch(List<? extends Row> actions, Object[] results)
528           throws IOException, InterruptedException {
529         table.batch(actions, results);
530       }
531 
532       /**
533        * {@inheritDoc}
534        * @deprecated If any exception is thrown by one of the actions, there is no way to
535        * retrieve the partially executed results. Use {@link #batch(List, Object[])} instead.
536        */
537       @Override
538       public Object[] batch(List<? extends Row> actions)
539           throws IOException, InterruptedException {
540         return table.batch(actions);
541       }
542 
543       @Override
544       public <R> void batchCallback(List<? extends Row> actions, Object[] results,
545           Batch.Callback<R> callback) throws IOException, InterruptedException {
546         table.batchCallback(actions, results, callback);
547       }
548 
549       /**
550        * {@inheritDoc}
551        * @deprecated If any exception is thrown by one of the actions, there is no way to
552        * retrieve the partially executed results. Use 
553        * {@link #batchCallback(List, Object[], org.apache.hadoop.hbase.client.coprocessor.Batch.Callback)}
554        * instead.
555        */
556       @Override
557       public <R> Object[] batchCallback(List<? extends Row> actions,
558           Batch.Callback<R> callback) throws IOException, InterruptedException {
559         return table.batchCallback(actions, callback);
560       }
561 
562       @Override
563       public Result[] get(List<Get> gets) throws IOException {
564         return table.get(gets);
565       }
566 
567       @Override
568       public CoprocessorRpcChannel coprocessorService(byte[] row) {
569         return table.coprocessorService(row);
570       }
571 
572       @Override
573       public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service,
574           byte[] startKey, byte[] endKey, Batch.Call<T, R> callable)
575           throws ServiceException, Throwable {
576         return table.coprocessorService(service, startKey, endKey, callable);
577       }
578 
579       @Override
580       public <T extends Service, R> void coprocessorService(Class<T> service,
581           byte[] startKey, byte[] endKey, Batch.Call<T, R> callable, Batch.Callback<R> callback)
582           throws ServiceException, Throwable {
583         table.coprocessorService(service, startKey, endKey, callable, callback);
584       }
585 
586       @Override
587       public void mutateRow(RowMutations rm) throws IOException {
588         table.mutateRow(rm);
589       }
590 
591       @Override
592       public void setAutoFlush(boolean autoFlush) {
593         table.setAutoFlush(autoFlush, autoFlush);
594       }
595 
596       @Override
597       public void setAutoFlush(boolean autoFlush, boolean clearBufferOnFail) {
598         table.setAutoFlush(autoFlush, clearBufferOnFail);
599       }
600 
601       @Override
602       public void setAutoFlushTo(boolean autoFlush) {
603         table.setAutoFlushTo(autoFlush);
604       }
605 
606       @Override
607       public long getWriteBufferSize() {
608          return table.getWriteBufferSize();
609       }
610 
611       @Override
612       public void setWriteBufferSize(long writeBufferSize) throws IOException {
613         table.setWriteBufferSize(writeBufferSize);
614       }
615 
616       @Override
617       public long incrementColumnValue(byte[] row, byte[] family,
618           byte[] qualifier, long amount, boolean writeToWAL) throws IOException {
619         return table.incrementColumnValue(row, family, qualifier, amount, writeToWAL);
620       }
621 
622       @Override
623       public <R extends Message> Map<byte[], R> batchCoprocessorService(
624           Descriptors.MethodDescriptor method, Message request, byte[] startKey,
625           byte[] endKey, R responsePrototype) throws ServiceException, Throwable {
626         return table.batchCoprocessorService(method, request, startKey, endKey, responsePrototype);
627       }
628 
629       @Override
630       public <R extends Message> void batchCoprocessorService(Descriptors.MethodDescriptor method,
631           Message request, byte[] startKey, byte[] endKey, R responsePrototype,
632           Callback<R> callback) throws ServiceException, Throwable {
633         table.batchCoprocessorService(method, request, startKey, endKey, responsePrototype,
634             callback);
635       }
636     }
637 
638     /** The coprocessor */
639     public Coprocessor impl;
640     /** Chaining priority */
641     protected int priority = Coprocessor.PRIORITY_USER;
642     /** Current coprocessor state */
643     Coprocessor.State state = Coprocessor.State.UNINSTALLED;
644     /** Accounting for tables opened by the coprocessor */
645     protected List<HTableInterface> openTables =
646       Collections.synchronizedList(new ArrayList<HTableInterface>());
647     private int seq;
648     private Configuration conf;
649 
650     /**
651      * Constructor
652      * @param impl the coprocessor instance
653      * @param priority chaining priority
654      */
655     public Environment(final Coprocessor impl, final int priority,
656         final int seq, final Configuration conf) {
657       this.impl = impl;
658       this.priority = priority;
659       this.state = Coprocessor.State.INSTALLED;
660       this.seq = seq;
661       this.conf = conf;
662     }
663 
664     /** Initialize the environment */
665     public void startup() throws IOException {
666       if (state == Coprocessor.State.INSTALLED ||
667           state == Coprocessor.State.STOPPED) {
668         state = Coprocessor.State.STARTING;
669         Thread currentThread = Thread.currentThread();
670         ClassLoader hostClassLoader = currentThread.getContextClassLoader();
671         try {
672           currentThread.setContextClassLoader(this.getClassLoader());
673           impl.start(this);
674           state = Coprocessor.State.ACTIVE;
675         } finally {
676           currentThread.setContextClassLoader(hostClassLoader);
677         }
678       } else {
679         LOG.warn("Not starting coprocessor "+impl.getClass().getName()+
680             " because not inactive (state="+state.toString()+")");
681       }
682     }
683 
684     /** Clean up the environment */
685     protected void shutdown() {
686       if (state == Coprocessor.State.ACTIVE) {
687         state = Coprocessor.State.STOPPING;
688         Thread currentThread = Thread.currentThread();
689         ClassLoader hostClassLoader = currentThread.getContextClassLoader();
690         try {
691           currentThread.setContextClassLoader(this.getClassLoader());
692           impl.stop(this);
693           state = Coprocessor.State.STOPPED;
694         } catch (IOException ioe) {
695           LOG.error("Error stopping coprocessor "+impl.getClass().getName(), ioe);
696         } finally {
697           currentThread.setContextClassLoader(hostClassLoader);
698         }
699       } else {
700         LOG.warn("Not stopping coprocessor "+impl.getClass().getName()+
701             " because not active (state="+state.toString()+")");
702       }
703       // clean up any table references
704       for (HTableInterface table: openTables) {
705         try {
706           ((HTableWrapper)table).internalClose();
707         } catch (IOException e) {
708           // nothing can be done here
709           LOG.warn("Failed to close " +
710               Bytes.toStringBinary(table.getTableName()), e);
711         }
712       }
713     }
714 
715     @Override
716     public Coprocessor getInstance() {
717       return impl;
718     }
719 
720     @Override
721     public ClassLoader getClassLoader() {
722       return impl.getClass().getClassLoader();
723     }
724 
725     @Override
726     public int getPriority() {
727       return priority;
728     }
729 
730     @Override
731     public int getLoadSequence() {
732       return seq;
733     }
734 
735     /** @return the coprocessor environment version */
736     @Override
737     public int getVersion() {
738       return Coprocessor.VERSION;
739     }
740 
741     /** @return the HBase release */
742     @Override
743     public String getHBaseVersion() {
744       return VersionInfo.getVersion();
745     }
746 
747     @Override
748     public Configuration getConfiguration() {
749       return conf;
750     }
751 
752     /**
753      * Open a table from within the Coprocessor environment
754      * @param tableName the table name
755      * @return an interface for manipulating the table
756      * @exception java.io.IOException Exception
757      */
758     @Override
759     public HTableInterface getTable(TableName tableName) throws IOException {
760       return this.getTable(tableName, HTable.getDefaultExecutor(getConfiguration()));
761     }
762 
763     /**
764      * Open a table from within the Coprocessor environment
765      * @param tableName the table name
766      * @return an interface for manipulating the table
767      * @exception java.io.IOException Exception
768      */
769     @Override
770     public HTableInterface getTable(TableName tableName, ExecutorService pool) throws IOException {
771       return new HTableWrapper(tableName, CoprocessorHConnection.getConnectionForEnvironment(this),
772           pool);
773     }
774   }
775 
776   protected void abortServer(final CoprocessorEnvironment environment, final Throwable e) {
777     abortServer(environment.getInstance().getClass().getName(), e);
778   }
779 
780   protected void abortServer(final String coprocessorName, final Throwable e) {
781     String message = "The coprocessor " + coprocessorName + " threw an unexpected exception";
782     LOG.error(message, e);
783     if (abortable != null) {
784       abortable.abort(message, e);
785     } else {
786       LOG.warn("No available Abortable, process was not aborted");
787     }
788   }
789 
790   /**
791    * This is used by coprocessor hooks which are declared to throw IOException
792    * (or its subtypes). For such hooks, we should handle throwable objects
793    * depending on the Throwable's type. Those which are instances of
794    * IOException should be passed on to the client. This is in conformance with
795    * the HBase idiom regarding IOException: that it represents a circumstance
796    * that should be passed along to the client for its own handling. For
797    * example, a coprocessor that implements access controls would throw a
798    * subclass of IOException, such as AccessDeniedException, in its preGet()
799    * method to prevent an unauthorized client's performing a Get on a particular
800    * table.
801    * @param env Coprocessor Environment
802    * @param e Throwable object thrown by coprocessor.
803    * @exception IOException Exception
804    */
805   protected void handleCoprocessorThrowable(final CoprocessorEnvironment env, final Throwable e)
806       throws IOException {
807     if (e instanceof IOException) {
808       throw (IOException)e;
809     }
810     // If we got here, e is not an IOException. A loaded coprocessor has a
811     // fatal bug, and the server (master or regionserver) should remove the
812     // faulty coprocessor from its set of active coprocessors. Setting
813     // 'hbase.coprocessor.abortonerror' to true will cause abortServer(),
814     // which may be useful in development and testing environments where
815     // 'failing fast' for error analysis is desired.
816     if (env.getConfiguration().getBoolean(ABORT_ON_ERROR_KEY, DEFAULT_ABORT_ON_ERROR)) {
817       // server is configured to abort.
818       abortServer(env, e);
819     } else {
820       LOG.error("Removing coprocessor '" + env.toString() + "' from " +
821           "environment because it threw:  " + e,e);
822       coprocessors.remove(env);
823       try {
824         shutdown(env);
825       } catch (Exception x) {
826         LOG.error("Uncaught exception when shutting down coprocessor '"
827             + env.toString() + "'", x);
828       }
829       throw new DoNotRetryIOException("Coprocessor: '" + env.toString() +
830           "' threw: '" + e + "' and has been removed from the active " +
831           "coprocessor set.", e);
832     }
833   }
834 }