HDDS-1893. Fix bug in removeAcl in Bucket. (#1216)

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java

@@ -542,6 +542,38 @@ public abstract class TestOzoneRpcClientAbstract {
     Assert.assertTrue(!bucket.getAcls().contains(acls.get(0)));
   }
 
+  @Test
+  public void testRemoveBucketAclUsingRpcClientRemoveAcl()
+      throws IOException {
+    String volumeName = UUID.randomUUID().toString();
+    String bucketName = UUID.randomUUID().toString();
+    OzoneAcl userAcl = new OzoneAcl(USER, "test",
+        ACLType.ALL, ACCESS);
+    List<OzoneAcl> acls = new ArrayList<>();
+    acls.add(userAcl);
+    acls.add(new OzoneAcl(USER, "test1",
+        ACLType.ALL, ACCESS));
+    store.createVolume(volumeName);
+    OzoneVolume volume = store.getVolume(volumeName);
+    BucketArgs.Builder builder = BucketArgs.newBuilder();
+    builder.setAcls(acls);
+    volume.createBucket(bucketName, builder.build());
+    OzoneObj ozoneObj = OzoneObjInfo.Builder.newBuilder()
+        .setBucketName(bucketName)
+        .setVolumeName(volumeName)
+        .setStoreType(OzoneObj.StoreType.OZONE)
+        .setResType(OzoneObj.ResourceType.BUCKET).build();
+
+    // Remove the 2nd acl added to the list.
+    boolean remove = store.removeAcl(ozoneObj, acls.get(1));
+    Assert.assertTrue(remove);
+    Assert.assertFalse(store.getAcl(ozoneObj).contains(acls.get(1)));
+
+    remove = store.removeAcl(ozoneObj, acls.get(0));
+    Assert.assertTrue(remove);
+    Assert.assertFalse(store.getAcl(ozoneObj).contains(acls.get(0)));
+  }
+
   @Test
   public void testSetBucketVersioning()
       throws IOException, OzoneException {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/BucketManagerImpl.java

@@ -505,6 +505,7 @@ public class BucketManagerImpl implements BucketManager {
             BUCKET_NOT_FOUND);
       }
 
+      boolean removed = false;
       // When we are removing subset of rights from existing acl.
       for(OzoneAcl a: bucketInfo.getAcls()) {
         if(a.getName().equals(acl.getName()) &&
@@ -515,20 +516,21 @@ public class BucketManagerImpl implements BucketManager {
           if (bits.equals(ZERO_BITSET)) {
             return false;
           }
-          bits = (BitSet) acl.getAclBitSet().clone();
+
-          bits.and(a.getAclBitSet());
           a.getAclBitSet().xor(bits);
 
           if(a.getAclBitSet().equals(ZERO_BITSET)) {
             bucketInfo.getAcls().remove(a);
           }
+          removed = true;
           break;
-        } else {
-          return false;
         }
       }
 
+      if (removed) {
         metadataManager.getBucketTable().put(dbBucketKey, bucketInfo);
+      }
+      return removed;
     } catch (IOException ex) {
       if (!(ex instanceof OMException)) {
         LOG.error("Remove acl operation failed for bucket:{}/{} acl:{}",
@@ -538,8 +540,6 @@ public class BucketManagerImpl implements BucketManager {
     } finally {
       metadataManager.getLock().releaseLock(BUCKET_LOCK, volume, bucket);
     }
-
-    return true;
   }
 
   /**