HDDS-1541. Implement addAcl,removeAcl,setAcl,getAcl for Key. Contributed by Ajay Kumat.

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -405,15 +405,7 @@ public void createBucket(
           .setKeyName(bucketArgs.getEncryptionKey()).build();
     }
 
-    List<OzoneAcl> listOfAcls = new ArrayList<>();
-    //User ACL
-    listOfAcls.add(new OzoneAcl(ACLIdentityType.USER,
-        ugi.getUserName(), userRights));
-    //Group ACLs of the User
-    List<String> userGroups = Arrays.asList(UserGroupInformation
-        .createRemoteUser(ugi.getUserName()).getGroupNames());
-    userGroups.stream().forEach((group) -> listOfAcls.add(
-        new OzoneAcl(ACLIdentityType.GROUP, group, groupRights)));
+    List<OzoneAcl> listOfAcls = getAclList();
     //ACLs from BucketArgs
     if(bucketArgs.getAcls() != null) {
       listOfAcls.addAll(bucketArgs.getAcls());
@@ -437,6 +429,16 @@ public void createBucket(
     ozoneManagerClient.createBucket(builder.build());
   }
 
+  /**
+   * Helper function to get default acl list for current user.
+   *
+   * @return listOfAcls
+   * */
+  private List<OzoneAcl> getAclList() {
+    return OzoneUtils.getAclList(ugi.getUserName(), ugi.getGroups(),
+        userRights, groupRights);
+  }
+
   @Override
   public void addBucketAcls(
       String volumeName, String bucketName, List<OzoneAcl> addAcls)
@@ -629,6 +631,7 @@ public OzoneOutputStream createKey(
         .setType(HddsProtos.ReplicationType.valueOf(type.toString()))
         .setFactor(HddsProtos.ReplicationFactor.valueOf(factor.getValue()))
         .addAllMetadata(metadata)
+        .setAcls(getAclList())
         .build();
 
     OpenKeySession openKey = ozoneManagerClient.openKey(keyArgs);
@@ -819,6 +822,7 @@ public OmMultipartInfo initiateMultipartUpload(String volumeName,
         .setKeyName(keyName)
         .setType(HddsProtos.ReplicationType.valueOf(type.toString()))
         .setFactor(HddsProtos.ReplicationFactor.valueOf(factor.getValue()))
+        .setAcls(getAclList())
         .build();
     OmMultipartInfo multipartInfo = ozoneManagerClient
         .initiateMultipartUpload(keyArgs);
@@ -848,6 +852,7 @@ public OzoneOutputStream createMultipartKey(String volumeName,
         .setIsMultipartKey(true)
         .setMultipartUploadID(uploadID)
         .setMultipartUploadPartNumber(partNumber)
+        .setAcls(getAclList())
         .build();
 
     OpenKeySession openKey = ozoneManagerClient.openKey(keyArgs);
@@ -963,7 +968,10 @@ public OzoneFileStatus getOzoneFileStatus(String volumeName,
   public void createDirectory(String volumeName, String bucketName,
       String keyName) throws IOException {
     OmKeyArgs keyArgs = new OmKeyArgs.Builder().setVolumeName(volumeName)
-        .setBucketName(bucketName).setKeyName(keyName).build();
+        .setBucketName(bucketName)
+        .setKeyName(keyName)
+        .setAcls(getAclList())
+        .build();
     ozoneManagerClient.createDirectory(keyArgs);
   }
 
@@ -990,6 +998,7 @@ public OzoneOutputStream createFile(String volumeName, String bucketName,
         .setDataSize(size)
         .setType(HddsProtos.ReplicationType.valueOf(type.name()))
         .setFactor(HddsProtos.ReplicationFactor.valueOf(factor.getValue()))
+        .setAcls(getAclList())
         .build();
     OpenKeySession keySession =
         ozoneManagerClient.createFile(keyArgs, overWrite, recursive);

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OzoneAcl.java

@@ -46,9 +46,10 @@ public class OzoneAcl {
   private ACLIdentityType type;
   private String name;
   private BitSet aclBitSet;
+  public static final BitSet ZERO_BITSET = new BitSet(0);
 
   /**
-   * Constructor for OzoneAcl.
+   * Default constructor.
    */
   public OzoneAcl() {
   }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java

@@ -19,6 +19,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.audit.Auditable;
 
@@ -45,13 +46,15 @@ public final class OmKeyArgs implements Auditable {
   private final int multipartUploadPartNumber;
   private Map<String, String> metadata;
   private boolean refreshPipeline;
+  private List<OzoneAcl> acls;
 
   @SuppressWarnings("parameternumber")
   private OmKeyArgs(String volumeName, String bucketName, String keyName,
       long dataSize, ReplicationType type, ReplicationFactor factor,
       List<OmKeyLocationInfo> locationInfoList, boolean isMultipart,
       String uploadID, int partNumber,
-      Map<String, String> metadataMap, boolean refreshPipeline) {
+      Map<String, String> metadataMap, boolean refreshPipeline,
+      List<OzoneAcl> acls) {
     this.volumeName = volumeName;
     this.bucketName = bucketName;
     this.keyName = keyName;
@@ -64,6 +67,7 @@ private OmKeyArgs(String volumeName, String bucketName, String keyName,
     this.multipartUploadPartNumber = partNumber;
     this.metadata = metadataMap;
     this.refreshPipeline = refreshPipeline;
+    this.acls = acls;
   }
 
   public boolean getIsMultipartKey() {
@@ -86,6 +90,10 @@ public ReplicationFactor getFactor() {
     return factor;
   }
 
+  public List<OzoneAcl> getAcls() {
+    return acls;
+  }
+
   public String getVolumeName() {
     return volumeName;
   }
@@ -166,6 +174,7 @@ public static class Builder {
     private int multipartUploadPartNumber;
     private Map<String, String> metadata = new HashMap<>();
     private boolean refreshPipeline;
+    private List<OzoneAcl> acls;
 
     public Builder setVolumeName(String volume) {
       this.volumeName = volume;
@@ -202,6 +211,11 @@ public Builder setLocationInfoList(List<OmKeyLocationInfo> locationInfos) {
       return this;
     }
 
+    public Builder setAcls(List<OzoneAcl> listOfAcls) {
+      this.acls = listOfAcls;
+      return this;
+    }
+
     public Builder setIsMultipartKey(boolean isMultipart) {
       this.isMultipartKey = isMultipart;
       return this;
@@ -235,7 +249,7 @@ public Builder setRefreshPipeline(boolean refresh) {
     public OmKeyArgs build() {
       return new OmKeyArgs(volumeName, bucketName, keyName, dataSize, type,
           factor, locationInfoList, isMultipartKey, multipartUploadID,
-          multipartUploadPartNumber, metadata, refreshPipeline);
+          multipartUploadPartNumber, metadata, refreshPipeline, acls);
     }
 
   }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java

@@ -28,6 +28,7 @@
 import org.apache.hadoop.fs.FileEncryptionInfo;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyInfo;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo;
 import org.apache.hadoop.ozone.protocolPB.OMPBHelper;
 import org.apache.hadoop.util.Time;
 
@@ -50,6 +51,10 @@ public final class OmKeyInfo extends WithMetadata {
   private HddsProtos.ReplicationType type;
   private HddsProtos.ReplicationFactor factor;
   private FileEncryptionInfo encInfo;
+  /**
+   * ACL Information.
+   */
+  private List<OzoneAclInfo> acls;
 
   @SuppressWarnings("parameternumber")
   OmKeyInfo(String volumeName, String bucketName, String keyName,
@@ -58,7 +63,7 @@ public final class OmKeyInfo extends WithMetadata {
       HddsProtos.ReplicationType type,
       HddsProtos.ReplicationFactor factor,
       Map<String, String> metadata,
-      FileEncryptionInfo encInfo) {
+      FileEncryptionInfo encInfo, List<OzoneAclInfo> acls) {
     this.volumeName = volumeName;
     this.bucketName = bucketName;
     this.keyName = keyName;
@@ -81,6 +86,7 @@ public final class OmKeyInfo extends WithMetadata {
     this.type = type;
     this.metadata = metadata;
     this.encInfo = encInfo;
+    this.acls = acls;
   }
 
   public String getVolumeName() {
@@ -216,6 +222,10 @@ public FileEncryptionInfo getFileEncryptionInfo() {
     return encInfo;
   }
 
+  public List<OzoneAclInfo> getAcls() {
+    return acls;
+  }
+
   /**
    * Builder of OmKeyInfo.
    */
@@ -232,6 +242,7 @@ public static class Builder {
     private HddsProtos.ReplicationFactor factor;
     private Map<String, String> metadata;
     private FileEncryptionInfo encInfo;
+    private List<OzoneAclInfo> acls;
 
     public Builder() {
       this.metadata = new HashMap<>();
@@ -299,11 +310,16 @@ public Builder setFileEncryptionInfo(FileEncryptionInfo feInfo) {
       return this;
     }
 
+    public Builder setAcls(List<OzoneAclInfo> listOfAcls) {
+      this.acls = listOfAcls;
+      return this;
+    }
+
     public OmKeyInfo build() {
       return new OmKeyInfo(
           volumeName, bucketName, keyName, omKeyLocationInfoGroups,
           dataSize, creationTime, modificationTime, type, factor, metadata,
-          encInfo);
+          encInfo, acls);
     }
   }
 
@@ -327,6 +343,9 @@ public KeyInfo getProtobuf() {
     if (encInfo != null) {
       kb.setFileEncryptionInfo(OMPBHelper.convert(encInfo));
     }
+    if(acls != null) {
+      kb.addAllAcls(acls);
+    }
     return kb.build();
   }
 
@@ -345,7 +364,8 @@ public static OmKeyInfo getFromProtobuf(KeyInfo keyInfo) {
         keyInfo.getFactor(),
         KeyValueUtil.getFromProtobuf(keyInfo.getMetadataList()),
         keyInfo.hasFileEncryptionInfo() ? OMPBHelper.convert(keyInfo
-            .getFileEncryptionInfo()): null);
+            .getFileEncryptionInfo()): null,
+        keyInfo.getAclsList());
   }
 
   @Override

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmOzoneAclMap.java

@@ -36,6 +36,7 @@
 import java.util.HashMap;
 import java.util.Objects;
 
+import static org.apache.hadoop.ozone.OzoneAcl.ZERO_BITSET;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_REQUEST;
 import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclRights.ALL;
 
@@ -81,8 +82,17 @@ public void addAcl(OzoneAcl acl) throws OMException {
     if (!getMap(aclType).containsKey(acl.getName())) {
       getMap(aclType).put(acl.getName(), acl.getAclBitSet());
     } else {
-      // throw exception if acl is already added.
-      throw new OMException("Acl " + acl + " already exist.", INVALID_REQUEST);
+      // Check if we are adding new rights to existing acl.
+      BitSet temp = (BitSet) acl.getAclBitSet().clone();
+      BitSet curRights = (BitSet) getMap(aclType).get(acl.getName()).clone();
+      temp.or(curRights);
+
+      if (temp.equals(curRights)) {
+        // throw exception if acl is already added.
+        throw new OMException("Acl " + acl + " already exist.",
+            INVALID_REQUEST);
+      }
+      getMap(aclType).get(acl.getName()).or(acl.getAclBitSet());
     }
   }
 
@@ -105,9 +115,25 @@ public void removeAcl(OzoneAcl acl) throws OMException {
     Objects.requireNonNull(acl, "Acl should not be null.");
     OzoneAclType aclType = OzoneAclType.valueOf(acl.getType().name());
     if (getMap(aclType).containsKey(acl.getName())) {
-      getMap(aclType).remove(acl.getName());
+      BitSet aclRights = getMap(aclType).get(acl.getName());
+      BitSet bits = (BitSet) acl.getAclBitSet().clone();
+      bits.and(aclRights);
+
+      if (bits.equals(ZERO_BITSET)) {
+        // throw exception if acl doesn't exist.
+        throw new OMException("Acl [" + acl + "] doesn't exist.",
+            INVALID_REQUEST);
+      }
+
+      acl.getAclBitSet().and(aclRights);
+      aclRights.xor(acl.getAclBitSet());
+
+      // Remove the acl as all rights are already set to 0.
+      if (aclRights.equals(ZERO_BITSET)) {
+        getMap(aclType).remove(acl.getName());
+      }
     } else {
-      // throw exception if acl is already added.
+      // throw exception if acl doesn't exist.
       throw new OMException("Acl [" + acl + "] doesn't exist.",
           INVALID_REQUEST);
     }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -668,6 +668,11 @@ public OpenKeySession openKey(OmKeyArgs args) throws IOException {
         .setBucketName(args.getBucketName())
         .setKeyName(args.getKeyName());
 
+    if(args.getAcls() != null) {
+      keyArgs.addAllAcls(args.getAcls().stream().distinct().map(a ->
+          OzoneAcl.toProtobuf(a)).collect(Collectors.toList()));
+    }
+
     if (args.getFactor() != null) {
       keyArgs.setFactor(args.getFactor());
     }
@@ -991,6 +996,8 @@ public OmMultipartInfo initiateMultipartUpload(OmKeyArgs omKeyArgs) throws
         .setBucketName(omKeyArgs.getBucketName())
         .setKeyName(omKeyArgs.getKeyName())
         .setFactor(omKeyArgs.getFactor())
+        .addAllAcls(omKeyArgs.getAcls().stream().map(a ->
+            OzoneAcl.toProtobuf(a)).collect(Collectors.toList()))
         .setType(omKeyArgs.getType());
     multipartInfoInitiateRequest.setKeyArgs(keyArgs.build());
 
@@ -1276,6 +1283,8 @@ public void createDirectory(OmKeyArgs args) throws IOException {
         .setVolumeName(args.getVolumeName())
         .setBucketName(args.getBucketName())
         .setKeyName(args.getKeyName())
+        .addAllAcls(args.getAcls().stream().map(a ->
+            OzoneAcl.toProtobuf(a)).collect(Collectors.toList()))
         .build();
     CreateDirectoryRequest request = CreateDirectoryRequest.newBuilder()
         .setKeyArgs(keyArgs)
@@ -1412,6 +1421,8 @@ public OpenKeySession createFile(OmKeyArgs args,
         .setDataSize(args.getDataSize())
         .setType(args.getType())
         .setFactor(args.getFactor())
+        .addAllAcls(args.getAcls().stream().map(a ->
+            OzoneAcl.toProtobuf(a)).collect(Collectors.toList()))
         .build();
     CreateFileRequest createFileRequest = CreateFileRequest.newBuilder()
             .setKeyArgs(keyArgs)