View Javadoc

1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.hbase.security.access;
20  
21  import java.io.Closeable;
22  import java.io.IOException;
23  import java.util.HashMap;
24  import java.util.List;
25  import java.util.Map;
26  import java.util.concurrent.ConcurrentSkipListMap;
27  import java.util.concurrent.atomic.AtomicLong;
28  
29  import org.apache.commons.logging.Log;
30  import org.apache.commons.logging.LogFactory;
31  import org.apache.hadoop.hbase.AuthUtil;
32  import org.apache.hadoop.hbase.classification.InterfaceAudience;
33  import org.apache.hadoop.conf.Configuration;
34  import org.apache.hadoop.hbase.Cell;
35  import org.apache.hadoop.hbase.TableName;
36  import org.apache.hadoop.hbase.exceptions.DeserializationException;
37  import org.apache.hadoop.hbase.security.Superusers;
38  import org.apache.hadoop.hbase.security.User;
39  import org.apache.hadoop.hbase.security.UserProvider;
40  import org.apache.hadoop.hbase.util.Bytes;
41  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
42  import org.apache.zookeeper.KeeperException;
43  
44  import com.google.common.annotations.VisibleForTesting;
45  import com.google.common.collect.ArrayListMultimap;
46  import com.google.common.collect.ListMultimap;
47  import com.google.common.collect.Lists;
48  
49  /**
50   * Performs authorization checks for a given user's assigned permissions
51   */
52  @InterfaceAudience.Private
53  public class TableAuthManager implements Closeable {
54    private static class PermissionCache<T extends Permission> {
55      /** Cache of user permissions */
56      private ListMultimap<String,T> userCache = ArrayListMultimap.create();
57      /** Cache of group permissions */
58      private ListMultimap<String,T> groupCache = ArrayListMultimap.create();
59  
60      public List<T> getUser(String user) {
61        return userCache.get(user);
62      }
63  
64      public void putUser(String user, T perm) {
65        userCache.put(user, perm);
66      }
67  
68      public List<T> replaceUser(String user, Iterable<? extends T> perms) {
69        return userCache.replaceValues(user, perms);
70      }
71  
72      public List<T> getGroup(String group) {
73        return groupCache.get(group);
74      }
75  
76      public void putGroup(String group, T perm) {
77        groupCache.put(group, perm);
78      }
79  
80      public List<T> replaceGroup(String group, Iterable<? extends T> perms) {
81        return groupCache.replaceValues(group, perms);
82      }
83  
84      /**
85       * Returns a combined map of user and group permissions, with group names prefixed by
86       * {@link AuthUtil#GROUP_PREFIX}.
87       */
88      public ListMultimap<String,T> getAllPermissions() {
89        ListMultimap<String,T> tmp = ArrayListMultimap.create();
90        tmp.putAll(userCache);
91        for (String group : groupCache.keySet()) {
92          tmp.putAll(AuthUtil.toGroupEntry(group), groupCache.get(group));
93        }
94        return tmp;
95      }
96    }
97  
98    private static final Log LOG = LogFactory.getLog(TableAuthManager.class);
99  
100   /** Cache of global permissions */
101   private volatile PermissionCache<Permission> globalCache;
102 
103   private ConcurrentSkipListMap<TableName, PermissionCache<TablePermission>> tableCache =
104       new ConcurrentSkipListMap<TableName, PermissionCache<TablePermission>>();
105 
106   private ConcurrentSkipListMap<String, PermissionCache<TablePermission>> nsCache =
107     new ConcurrentSkipListMap<String, PermissionCache<TablePermission>>();
108 
109   private Configuration conf;
110   private ZKPermissionWatcher zkperms;
111   private final AtomicLong mtime = new AtomicLong(0L);
112 
113   private TableAuthManager(ZooKeeperWatcher watcher, Configuration conf)
114       throws IOException {
115     this.conf = conf;
116 
117     // initialize global permissions based on configuration
118     globalCache = initGlobal(conf);
119 
120     this.zkperms = new ZKPermissionWatcher(watcher, this, conf);
121     try {
122       this.zkperms.start();
123     } catch (KeeperException ke) {
124       LOG.error("ZooKeeper initialization failed", ke);
125     }
126   }
127 
128   @Override
129   public void close() {
130     this.zkperms.close();
131   }
132 
133   /**
134    * Returns a new {@code PermissionCache} initialized with permission assignments
135    * from the {@code hbase.superuser} configuration key.
136    */
137   private PermissionCache<Permission> initGlobal(Configuration conf) throws IOException {
138     UserProvider userProvider = UserProvider.instantiate(conf);
139     User user = userProvider.getCurrent();
140     if (user == null) {
141       throw new IOException("Unable to obtain the current user, " +
142           "authorization checks for internal operations will not work correctly!");
143     }
144     PermissionCache<Permission> newCache = new PermissionCache<Permission>();
145     String currentUser = user.getShortName();
146 
147     // the system user is always included
148     List<String> superusers = Lists.asList(currentUser, conf.getStrings(
149         Superusers.SUPERUSER_CONF_KEY, new String[0]));
150     if (superusers != null) {
151       for (String name : superusers) {
152         if (AuthUtil.isGroupPrincipal(name)) {
153           newCache.putGroup(AuthUtil.getGroupName(name),
154               new Permission(Permission.Action.values()));
155         } else {
156           newCache.putUser(name, new Permission(Permission.Action.values()));
157         }
158       }
159     }
160     return newCache;
161   }
162 
163   public ZKPermissionWatcher getZKPermissionWatcher() {
164     return this.zkperms;
165   }
166 
167   public void refreshTableCacheFromWritable(TableName table,
168                                        byte[] data) throws IOException {
169     if (data != null && data.length > 0) {
170       ListMultimap<String,TablePermission> perms;
171       try {
172         perms = AccessControlLists.readPermissions(data, conf);
173       } catch (DeserializationException e) {
174         throw new IOException(e);
175       }
176 
177       if (perms != null) {
178         if (Bytes.equals(table.getName(), AccessControlLists.ACL_GLOBAL_NAME)) {
179           updateGlobalCache(perms);
180         } else {
181           updateTableCache(table, perms);
182         }
183       }
184     } else {
185       LOG.debug("Skipping permission cache refresh because writable data is empty");
186     }
187   }
188 
189   public void refreshNamespaceCacheFromWritable(String namespace, byte[] data) throws IOException {
190     if (data != null && data.length > 0) {
191       ListMultimap<String,TablePermission> perms;
192       try {
193         perms = AccessControlLists.readPermissions(data, conf);
194       } catch (DeserializationException e) {
195         throw new IOException(e);
196       }
197       if (perms != null) {
198         updateNsCache(namespace, perms);
199       }
200     } else {
201       LOG.debug("Skipping permission cache refresh because writable data is empty");
202     }
203   }
204 
205   /**
206    * Updates the internal global permissions cache
207    *
208    * @param userPerms
209    */
210   private void updateGlobalCache(ListMultimap<String,TablePermission> userPerms) {
211     PermissionCache<Permission> newCache = null;
212     try {
213       newCache = initGlobal(conf);
214       for (Map.Entry<String,TablePermission> entry : userPerms.entries()) {
215         if (AuthUtil.isGroupPrincipal(entry.getKey())) {
216           newCache.putGroup(AuthUtil.getGroupName(entry.getKey()),
217               new Permission(entry.getValue().getActions()));
218         } else {
219           newCache.putUser(entry.getKey(), new Permission(entry.getValue().getActions()));
220         }
221       }
222       globalCache = newCache;
223       mtime.incrementAndGet();
224     } catch (IOException e) {
225       // Never happens
226       LOG.error("Error occured while updating the global cache", e);
227     }
228   }
229 
230   /**
231    * Updates the internal permissions cache for a single table, splitting
232    * the permissions listed into separate caches for users and groups to optimize
233    * group lookups.
234    *
235    * @param table
236    * @param tablePerms
237    */
238   private void updateTableCache(TableName table,
239                                 ListMultimap<String,TablePermission> tablePerms) {
240     PermissionCache<TablePermission> newTablePerms = new PermissionCache<TablePermission>();
241 
242     for (Map.Entry<String,TablePermission> entry : tablePerms.entries()) {
243       if (AuthUtil.isGroupPrincipal(entry.getKey())) {
244         newTablePerms.putGroup(AuthUtil.getGroupName(entry.getKey()), entry.getValue());
245       } else {
246         newTablePerms.putUser(entry.getKey(), entry.getValue());
247       }
248     }
249 
250     tableCache.put(table, newTablePerms);
251     mtime.incrementAndGet();
252   }
253 
254   /**
255    * Updates the internal permissions cache for a single table, splitting
256    * the permissions listed into separate caches for users and groups to optimize
257    * group lookups.
258    *
259    * @param namespace
260    * @param tablePerms
261    */
262   private void updateNsCache(String namespace,
263                              ListMultimap<String, TablePermission> tablePerms) {
264     PermissionCache<TablePermission> newTablePerms = new PermissionCache<TablePermission>();
265 
266     for (Map.Entry<String, TablePermission> entry : tablePerms.entries()) {
267       if (AuthUtil.isGroupPrincipal(entry.getKey())) {
268         newTablePerms.putGroup(AuthUtil.getGroupName(entry.getKey()), entry.getValue());
269       } else {
270         newTablePerms.putUser(entry.getKey(), entry.getValue());
271       }
272     }
273 
274     nsCache.put(namespace, newTablePerms);
275     mtime.incrementAndGet();
276   }
277 
278   private PermissionCache<TablePermission> getTablePermissions(TableName table) {
279     if (!tableCache.containsKey(table)) {
280       tableCache.putIfAbsent(table, new PermissionCache<TablePermission>());
281     }
282     return tableCache.get(table);
283   }
284 
285   private PermissionCache<TablePermission> getNamespacePermissions(String namespace) {
286     if (!nsCache.containsKey(namespace)) {
287       nsCache.putIfAbsent(namespace, new PermissionCache<TablePermission>());
288     }
289     return nsCache.get(namespace);
290   }
291 
292   /**
293    * Authorizes a global permission
294    * @param perms
295    * @param action
296    * @return true if authorized, false otherwise
297    */
298   private boolean authorize(List<Permission> perms, Permission.Action action) {
299     if (perms != null) {
300       for (Permission p : perms) {
301         if (p.implies(action)) {
302           return true;
303         }
304       }
305     } else if (LOG.isDebugEnabled()) {
306       LOG.debug("No permissions found for " + action);
307     }
308 
309     return false;
310   }
311 
312   /**
313    * Authorize a global permission based on ACLs for the given user and the
314    * user's groups.
315    * @param user
316    * @param action
317    * @return true if known and authorized, false otherwise
318    */
319   public boolean authorize(User user, Permission.Action action) {
320     if (user == null) {
321       return false;
322     }
323 
324     if (authorize(globalCache.getUser(user.getShortName()), action)) {
325       return true;
326     }
327 
328     String[] groups = user.getGroupNames();
329     if (groups != null) {
330       for (String group : groups) {
331         if (authorize(globalCache.getGroup(group), action)) {
332           return true;
333         }
334       }
335     }
336     return false;
337   }
338 
339   private boolean authorize(List<TablePermission> perms,
340                             TableName table, byte[] family,
341                             byte[] qualifier, Permission.Action action) {
342     if (perms != null) {
343       for (TablePermission p : perms) {
344         if (p.implies(table, family, qualifier, action)) {
345           return true;
346         }
347       }
348     } else if (LOG.isDebugEnabled()) {
349       LOG.debug("No permissions found for table="+table);
350     }
351     return false;
352   }
353 
354   private boolean hasAccess(List<TablePermission> perms,
355                             TableName table, Permission.Action action) {
356     if (perms != null) {
357       for (TablePermission p : perms) {
358         if (p.implies(action)) {
359           return true;
360         }
361       }
362     } else if (LOG.isDebugEnabled()) {
363       LOG.debug("No permissions found for table="+table);
364     }
365     return false;
366   }
367 
368   /**
369    * Authorize a user for a given KV. This is called from AccessControlFilter.
370    */
371   public boolean authorize(User user, TableName table, Cell cell, Permission.Action action) {
372     try {
373       List<Permission> perms = AccessControlLists.getCellPermissionsForUser(user, cell);
374       if (LOG.isTraceEnabled()) {
375         LOG.trace("Perms for user " + user.getShortName() + " in cell " + cell + ": " +
376           (perms != null ? perms : ""));
377       }
378       if (perms != null) {
379         for (Permission p: perms) {
380           if (p.implies(action)) {
381             return true;
382           }
383         }
384       }
385     } catch (IOException e) {
386       // We failed to parse the KV tag
387       LOG.error("Failed parse of ACL tag in cell " + cell);
388       // Fall through to check with the table and CF perms we were able
389       // to collect regardless
390     }
391     return false;
392   }
393 
394   public boolean authorize(User user, String namespace, Permission.Action action) {
395     // Global authorizations supercede namespace level
396     if (authorize(user, action)) {
397       return true;
398     }
399     // Check namespace permissions
400     PermissionCache<TablePermission> tablePerms = nsCache.get(namespace);
401     if (tablePerms != null) {
402       List<TablePermission> userPerms = tablePerms.getUser(user.getShortName());
403       if (authorize(userPerms, namespace, action)) {
404         return true;
405       }
406       String[] groupNames = user.getGroupNames();
407       if (groupNames != null) {
408         for (String group : groupNames) {
409           List<TablePermission> groupPerms = tablePerms.getGroup(group);
410           if (authorize(groupPerms, namespace, action)) {
411             return true;
412           }
413         }
414       }
415     }
416     return false;
417   }
418 
419   private boolean authorize(List<TablePermission> perms, String namespace,
420                             Permission.Action action) {
421     if (perms != null) {
422       for (TablePermission p : perms) {
423         if (p.implies(namespace, action)) {
424           return true;
425         }
426       }
427     } else if (LOG.isDebugEnabled()) {
428       LOG.debug("No permissions for authorize() check, table=" + namespace);
429     }
430 
431     return false;
432   }
433 
434   /**
435    * Checks authorization to a given table and column family for a user, based on the
436    * stored user permissions.
437    *
438    * @param user
439    * @param table
440    * @param family
441    * @param action
442    * @return true if known and authorized, false otherwise
443    */
444   public boolean authorizeUser(User user, TableName table, byte[] family,
445       Permission.Action action) {
446     return authorizeUser(user, table, family, null, action);
447   }
448 
449   public boolean authorizeUser(User user, TableName table, byte[] family,
450       byte[] qualifier, Permission.Action action) {
451     if (table == null) table = AccessControlLists.ACL_TABLE_NAME;
452     // Global and namespace authorizations supercede table level
453     if (authorize(user, table.getNamespaceAsString(), action)) {
454       return true;
455     }
456     // Check table permissions
457     return authorize(getTablePermissions(table).getUser(user.getShortName()), table, family,
458         qualifier, action);
459   }
460 
461   /**
462    * Checks if the user has access to the full table or at least a family/qualifier
463    * for the specified action.
464    *
465    * @param user
466    * @param table
467    * @param action
468    * @return true if the user has access to the table, false otherwise
469    */
470   public boolean userHasAccess(User user, TableName table, Permission.Action action) {
471     if (table == null) table = AccessControlLists.ACL_TABLE_NAME;
472     // Global and namespace authorizations supercede table level
473     if (authorize(user, table.getNamespaceAsString(), action)) {
474       return true;
475     }
476     // Check table permissions
477     return hasAccess(getTablePermissions(table).getUser(user.getShortName()), table, action);
478   }
479 
480   /**
481    * Checks global authorization for a given action for a group, based on the stored
482    * permissions.
483    */
484   public boolean authorizeGroup(String groupName, Permission.Action action) {
485     List<Permission> perms = globalCache.getGroup(groupName);
486     if (LOG.isDebugEnabled()) {
487       LOG.debug("authorizing " + (perms != null && !perms.isEmpty() ? perms.get(0) : "") +
488         " for " + action);
489     }
490     return authorize(perms, action);
491   }
492 
493   /**
494    * Checks authorization to a given table, column family and column for a group, based
495    * on the stored permissions.
496    * @param groupName
497    * @param table
498    * @param family
499    * @param qualifier
500    * @param action
501    * @return true if known and authorized, false otherwise
502    */
503   public boolean authorizeGroup(String groupName, TableName table, byte[] family,
504       byte[] qualifier, Permission.Action action) {
505     // Global authorization supercedes table level
506     if (authorizeGroup(groupName, action)) {
507       return true;
508     }
509     if (table == null) table = AccessControlLists.ACL_TABLE_NAME;
510     // Namespace authorization supercedes table level
511     String namespace = table.getNamespaceAsString();
512     if (authorize(getNamespacePermissions(namespace).getGroup(groupName), namespace, action)) {
513       return true;
514     }
515     // Check table level
516     List<TablePermission> tblPerms = getTablePermissions(table).getGroup(groupName);
517     if (LOG.isDebugEnabled()) {
518       LOG.debug("authorizing " + (tblPerms != null && !tblPerms.isEmpty() ? tblPerms.get(0) : "") +
519         " for " +groupName + " on " + table + "." + Bytes.toString(family) + "." +
520         Bytes.toString(qualifier) + " with " + action);
521     }
522     return authorize(tblPerms, table, family, qualifier, action);
523   }
524 
525   /**
526    * Checks if the user has access to the full table or at least a family/qualifier
527    * for the specified action.
528    * @param groupName
529    * @param table
530    * @param action
531    * @return true if the group has access to the table, false otherwise
532    */
533   public boolean groupHasAccess(String groupName, TableName table, Permission.Action action) {
534     // Global authorization supercedes table level
535     if (authorizeGroup(groupName, action)) {
536       return true;
537     }
538     if (table == null) table = AccessControlLists.ACL_TABLE_NAME;
539     // Namespace authorization supercedes table level
540     if (hasAccess(getNamespacePermissions(table.getNamespaceAsString()).getGroup(groupName),
541         table, action)) {
542       return true;
543     }
544     // Check table level
545     return hasAccess(getTablePermissions(table).getGroup(groupName), table, action);
546   }
547 
548   public boolean authorize(User user, TableName table, byte[] family,
549       byte[] qualifier, Permission.Action action) {
550     if (authorizeUser(user, table, family, qualifier, action)) {
551       return true;
552     }
553 
554     String[] groups = user.getGroupNames();
555     if (groups != null) {
556       for (String group : groups) {
557         if (authorizeGroup(group, table, family, qualifier, action)) {
558           return true;
559         }
560       }
561     }
562     return false;
563   }
564 
565   public boolean hasAccess(User user, TableName table, Permission.Action action) {
566     if (userHasAccess(user, table, action)) {
567       return true;
568     }
569 
570     String[] groups = user.getGroupNames();
571     if (groups != null) {
572       for (String group : groups) {
573         if (groupHasAccess(group, table, action)) {
574           return true;
575         }
576       }
577     }
578     return false;
579   }
580 
581   public boolean authorize(User user, TableName table, byte[] family,
582       Permission.Action action) {
583     return authorize(user, table, family, null, action);
584   }
585 
586   /**
587    * Returns true if the given user has a {@link TablePermission} matching up
588    * to the column family portion of a permission.  Note that this permission
589    * may be scoped to a given column qualifier and does not guarantee that
590    * authorize() on the same column family would return true.
591    */
592   public boolean matchPermission(User user,
593       TableName table, byte[] family, Permission.Action action) {
594     PermissionCache<TablePermission> tablePerms = tableCache.get(table);
595     if (tablePerms != null) {
596       List<TablePermission> userPerms = tablePerms.getUser(user.getShortName());
597       if (userPerms != null) {
598         for (TablePermission p : userPerms) {
599           if (p.matchesFamily(table, family, action)) {
600             return true;
601           }
602         }
603       }
604 
605       String[] groups = user.getGroupNames();
606       if (groups != null) {
607         for (String group : groups) {
608           List<TablePermission> groupPerms = tablePerms.getGroup(group);
609           if (groupPerms != null) {
610             for (TablePermission p : groupPerms) {
611               if (p.matchesFamily(table, family, action)) {
612                 return true;
613               }
614             }
615           }
616         }
617       }
618     }
619 
620     return false;
621   }
622 
623   public boolean matchPermission(User user,
624       TableName table, byte[] family, byte[] qualifier,
625       Permission.Action action) {
626     PermissionCache<TablePermission> tablePerms = tableCache.get(table);
627     if (tablePerms != null) {
628       List<TablePermission> userPerms = tablePerms.getUser(user.getShortName());
629       if (userPerms != null) {
630         for (TablePermission p : userPerms) {
631           if (p.matchesFamilyQualifier(table, family, qualifier, action)) {
632             return true;
633           }
634         }
635       }
636 
637       String[] groups = user.getGroupNames();
638       if (groups != null) {
639         for (String group : groups) {
640           List<TablePermission> groupPerms = tablePerms.getGroup(group);
641           if (groupPerms != null) {
642             for (TablePermission p : groupPerms) {
643               if (p.matchesFamilyQualifier(table, family, qualifier, action)) {
644                 return true;
645               }
646             }
647           }
648         }
649       }
650     }
651     return false;
652   }
653 
654   public void removeNamespace(byte[] ns) {
655     nsCache.remove(Bytes.toString(ns));
656   }
657 
658   public void removeTable(TableName table) {
659     tableCache.remove(table);
660   }
661 
662   /**
663    * Overwrites the existing permission set for a given user for a table, and
664    * triggers an update for zookeeper synchronization.
665    * @param username
666    * @param table
667    * @param perms
668    */
669   public void setTableUserPermissions(String username, TableName table,
670       List<TablePermission> perms) {
671     PermissionCache<TablePermission> tablePerms = getTablePermissions(table);
672     tablePerms.replaceUser(username, perms);
673     writeTableToZooKeeper(table, tablePerms);
674   }
675 
676   /**
677    * Overwrites the existing permission set for a group and triggers an update
678    * for zookeeper synchronization.
679    * @param group
680    * @param table
681    * @param perms
682    */
683   public void setTableGroupPermissions(String group, TableName table,
684       List<TablePermission> perms) {
685     PermissionCache<TablePermission> tablePerms = getTablePermissions(table);
686     tablePerms.replaceGroup(group, perms);
687     writeTableToZooKeeper(table, tablePerms);
688   }
689 
690   /**
691    * Overwrites the existing permission set for a given user for a table, and
692    * triggers an update for zookeeper synchronization.
693    * @param username
694    * @param namespace
695    * @param perms
696    */
697   public void setNamespaceUserPermissions(String username, String namespace,
698       List<TablePermission> perms) {
699     PermissionCache<TablePermission> tablePerms = getNamespacePermissions(namespace);
700     tablePerms.replaceUser(username, perms);
701     writeNamespaceToZooKeeper(namespace, tablePerms);
702   }
703 
704   /**
705    * Overwrites the existing permission set for a group and triggers an update
706    * for zookeeper synchronization.
707    * @param group
708    * @param namespace
709    * @param perms
710    */
711   public void setNamespaceGroupPermissions(String group, String namespace,
712       List<TablePermission> perms) {
713     PermissionCache<TablePermission> tablePerms = getNamespacePermissions(namespace);
714     tablePerms.replaceGroup(group, perms);
715     writeNamespaceToZooKeeper(namespace, tablePerms);
716   }
717 
718   public void writeTableToZooKeeper(TableName table,
719       PermissionCache<TablePermission> tablePerms) {
720     byte[] serialized = new byte[0];
721     if (tablePerms != null) {
722       serialized = AccessControlLists.writePermissionsAsBytes(tablePerms.getAllPermissions(), conf);
723     }
724     zkperms.writeToZookeeper(table.getName(), serialized);
725   }
726 
727   public void writeNamespaceToZooKeeper(String namespace,
728       PermissionCache<TablePermission> tablePerms) {
729     byte[] serialized = new byte[0];
730     if (tablePerms != null) {
731       serialized = AccessControlLists.writePermissionsAsBytes(tablePerms.getAllPermissions(), conf);
732     }
733     zkperms.writeToZookeeper(Bytes.toBytes(AccessControlLists.toNamespaceEntry(namespace)),
734         serialized);
735   }
736 
737   public long getMTime() {
738     return mtime.get();
739   }
740 
741   private static Map<ZooKeeperWatcher,TableAuthManager> managerMap =
742     new HashMap<ZooKeeperWatcher,TableAuthManager>();
743 
744   private static Map<TableAuthManager, Integer> refCount = new HashMap<>();
745 
746   /** Returns a TableAuthManager from the cache. If not cached, constructs a new one. Returned
747    * instance should be released back by calling {@link #release(TableAuthManager)}. */
748   public synchronized static TableAuthManager getOrCreate(
749       ZooKeeperWatcher watcher, Configuration conf) throws IOException {
750     TableAuthManager instance = managerMap.get(watcher);
751     if (instance == null) {
752       instance = new TableAuthManager(watcher, conf);
753       managerMap.put(watcher, instance);
754     }
755     int ref = refCount.get(instance) == null ? 0 : refCount.get(instance).intValue();
756     refCount.put(instance, ref + 1);
757     return instance;
758   }
759 
760   @VisibleForTesting
761   static int getTotalRefCount() {
762     int total = 0;
763     for (int count : refCount.values()) {
764       total += count;
765     }
766     return total;
767   }
768 
769   /**
770    * Releases the resources for the given TableAuthManager if the reference count is down to 0.
771    * @param instance TableAuthManager to be released
772    */
773   public synchronized static void release(TableAuthManager instance) {
774     if (refCount.get(instance) == null || refCount.get(instance) < 1) {
775       String msg = "Something wrong with the TableAuthManager reference counting: " + instance
776           + " whose count is " + refCount.get(instance);
777       LOG.fatal(msg);
778       instance.close();
779       managerMap.remove(instance.getZKPermissionWatcher().getWatcher());
780       instance.getZKPermissionWatcher().getWatcher().abort(msg, null);
781     } else {
782       int ref = refCount.get(instance);
783       refCount.put(instance, ref-1);
784       if (ref-1 == 0) {
785         instance.close();
786         managerMap.remove(instance.getZKPermissionWatcher().getWatcher());
787         refCount.remove(instance);
788       }
789     }
790   }
791 }