svn commit: r1552749 - /hbase/branches/0.98/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java

[an...@apache.org]

Author: anoopsamjohn
Date: Fri Dec 20 17:25:21 2013
New Revision: 1552749

URL: http://svn.apache.org/r1552749
Log:
HBASE-10161 [AccessController] Tolerate regions in recovery

Modified:
    hbase/branches/0.98/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java

Modified: hbase/branches/0.98/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.98/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java?rev=1552749&r1=1552748&r2=1552749&view=diff
==============================================================================
--- hbase/branches/0.98/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java (original)
+++ hbase/branches/0.98/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java Fri Dec 20 17:25:21 2013
@@ -153,6 +153,8 @@ public class AccessController extends Ba
   // flags if we are able to support cell ACLs
   boolean canPersistCellACLs;
 
+  private volatile boolean initialized = false;
+
   void initialize(RegionCoprocessorEnvironment e) throws IOException {
     final HRegion region = e.getRegion();
     Map<byte[], ListMultimap<String,TablePermission>> tables =
@@ -166,6 +168,7 @@ public class AccessController extends Ba
       byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms, e.getConfiguration());
       this.authManager.getZKPermissionWatcher().writeToZookeeper(entry, serialized);
     }
+    initialized = true;
   }
 
   /**
@@ -1043,8 +1046,26 @@ public class AccessController extends Ba
     }
     if (AccessControlLists.isAclRegion(region)) {
       aclRegion = true;
+      // When this region is under recovering state, initialize will be handled by postLogReplay
+      if (!region.isRecovering()) {
+        try {
+          initialize(env);
+        } catch (IOException ex) {
+          // if we can't obtain permissions, it's better to fail
+          // than perform checks incorrectly
+          throw new RuntimeException("Failed to initialize permissions cache", ex);
+        }
+      }
+    } else {
+      initialized = true;
+    }
+  }
+
+  @Override
+  public void postLogReplay(ObserverContext<RegionCoprocessorEnvironment> c) {
+    if (aclRegion) {
       try {
-        initialize(env);
+        initialize(c.getEnvironment());
       } catch (IOException ex) {
         // if we can't obtain permissions, it's better to fail
         // than perform checks incorrectly
@@ -1456,6 +1477,9 @@ public class AccessController extends Ba
     try {
       // verify it's only running at .acl.
       if (aclRegion) {
+        if (!initialized) {
+          throw new CoprocessorException("AccessController not yet initialized");
+        }
         if (LOG.isDebugEnabled()) {
           LOG.debug("Received request to grant access permission " + perm.toString());
         }
@@ -1496,6 +1520,9 @@ public class AccessController extends Ba
     try {
       // only allowed to be called on _acl_ region
       if (aclRegion) {
+        if (!initialized) {
+          throw new CoprocessorException("AccessController not yet initialized");
+        }
         if (LOG.isDebugEnabled()) {
           LOG.debug("Received request to revoke access permission " + perm.toString());
         }
@@ -1535,6 +1562,9 @@ public class AccessController extends Ba
     try {
       // only allowed to be called on _acl_ region
       if (aclRegion) {
+        if (!initialized) {
+          throw new CoprocessorException("AccessController not yet initialized");
+        }
         List<UserPermission> perms = null;
         if(request.getType() == AccessControlProtos.Permission.Type.Table) {
           TableName table = null;