HDDS-14122. Make OmKeyInfo ACL list immutable

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

@@ -24,7 +24,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Consumer;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
@@ -94,7 +93,7 @@ public final class OmKeyInfo extends WithParentObjectId
   /**
    * ACL Information.
    */
-  private final CopyOnWriteArrayList<OzoneAcl> acls;
+  private final ImmutableList<OzoneAcl> acls;
 
   /**
    * Used for S3 tags.
@@ -120,7 +119,7 @@ private OmKeyInfo(Builder b) {
     this.modificationTime = b.modificationTime;
     this.replicationConfig = b.replicationConfig;
     this.encInfo = b.encInfo;
-    this.acls = new CopyOnWriteArrayList<>(b.acls);
+    this.acls = b.acls.build();
     this.fileChecksum = b.fileChecksum;
     this.fileName = b.fileName;
     this.isFile = b.isFile;
@@ -452,19 +451,7 @@ public FileEncryptionInfo getFileEncryptionInfo() {
   }
 
   public List<OzoneAcl> getAcls() {
-    return ImmutableList.copyOf(acls);
-  }
-
-  public boolean addAcl(OzoneAcl acl) {
-    return OzoneAclUtil.addAcl(acls, acl);
-  }
-
-  public boolean removeAcl(OzoneAcl acl) {
-    return OzoneAclUtil.removeAcl(acls, acl);
-  }
-
-  public boolean setAcls(List<OzoneAcl> newAcls) {
-    return OzoneAclUtil.setAcl(acls, newAcls);
+    return acls;
   }
 
   public void setReplicationConfig(ReplicationConfig repConfig) {
@@ -509,7 +496,7 @@ public static class Builder extends WithParentObjectId.Builder<OmKeyInfo> {
     private long modificationTime;
     private ReplicationConfig replicationConfig;
     private FileEncryptionInfo encInfo;
-    private final List<OzoneAcl> acls = new ArrayList<>();
+    private final AclListBuilder acls;
     // not persisted to DB. FileName will be the last element in path keyName.
     private String fileName;
     private FileChecksum fileChecksum;
@@ -519,10 +506,16 @@ public static class Builder extends WithParentObjectId.Builder<OmKeyInfo> {
     private Long expectedDataGeneration = null;
 
     public Builder() {
+      this(AclListBuilder.empty());
+    }
+
+    private Builder(AclListBuilder acls) {
+      this.acls = acls;
     }
 
     public Builder(OmKeyInfo obj) {
       super(obj);
+      this.acls = AclListBuilder.of(obj.acls);
       this.volumeName = obj.volumeName;
       this.bucketName = obj.bucketName;
       this.keyName = obj.keyName;
@@ -539,7 +532,6 @@ public Builder(OmKeyInfo obj) {
       if (obj.getTags() != null) {
         this.tags.putAll(obj.getTags());
       }
-      this.acls.addAll(obj.getAcls());
       obj.keyLocationVersions.forEach(keyLocationVersion ->
           this.omKeyLocationInfoGroups.add(
               new OmKeyLocationInfoGroup(keyLocationVersion.getVersion(),
@@ -631,11 +623,15 @@ public Builder setFileEncryptionInfo(FileEncryptionInfo feInfo) {
 
     public Builder setAcls(List<OzoneAcl> listOfAcls) {
       if (listOfAcls != null) {
-        this.acls.addAll(listOfAcls);
+        this.acls.set(listOfAcls);
       }
       return this;
     }
 
+    public AclListBuilder acls() {
+      return acls;
+    }
+
     public Builder addAcl(OzoneAcl ozoneAcl) {
       if (ozoneAcl != null) {
         this.acls.add(ozoneAcl);

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmKeyInfo.java

@@ -189,9 +189,11 @@ private void createdAndTest(boolean isMPU) {
       }
     }
 
-    key.setAcls(Arrays.asList(OzoneAcl.of(
-        IAccessAuthorizer.ACLIdentityType.USER, "user1",
-        ACCESS, IAccessAuthorizer.ACLType.WRITE)));
+    key = key.toBuilder()
+        .setAcls(Arrays.asList(OzoneAcl.of(
+            IAccessAuthorizer.ACLIdentityType.USER, "user1",
+            ACCESS, IAccessAuthorizer.ACLType.WRITE)))
+        .build();
 
     // Change acls and check.
     assertNotEquals(key, cloneKey);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAclRequest.java

@@ -106,10 +106,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         throw new OMException(OMException.ResultCodes.KEY_NOT_FOUND);
       }
 
-      operationResult = apply(omKeyInfo, trxnLogIndex);
-      omKeyInfo = omKeyInfo.toBuilder()
-          .setUpdateID(trxnLogIndex)
-          .build();
+      OmKeyInfo.Builder builder = omKeyInfo.toBuilder();
+      operationResult = apply(builder, trxnLogIndex);
 
       // Update the modification time when updating ACLs of Key.
       long modificationTime = omKeyInfo.getModificationTime();
@@ -125,7 +123,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         modificationTime = getOmRequest().getRemoveAclRequest()
             .getModificationTime();
       }
-      omKeyInfo.setModificationTime(modificationTime);
+
+      omKeyInfo = builder
+          .setModificationTime(modificationTime)
+          .setUpdateID(trxnLogIndex)
+          .build();
 
       // update cache.
       omMetadataManager.getKeyTable(getBucketLayout())
@@ -244,10 +246,10 @@ abstract void onComplete(Result result, boolean operationResult,
   /**
    * Apply the acl operation, if successfully completed returns true,
    * else false.
-   * @param omKeyInfo
+   * @param builder
    * @param trxnLogIndex
    */
-  abstract boolean apply(OmKeyInfo omKeyInfo, long trxnLogIndex);
+  abstract boolean apply(OmKeyInfo.Builder builder, long trxnLogIndex);
 
   public void setBucketLayout(BucketLayout bucketLayout) {
     this.bucketLayout = bucketLayout;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAclRequestWithFSO.java

@@ -97,19 +97,17 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         throw new OMException("Key not found. Key:" + key, KEY_NOT_FOUND);
       }
       omKeyInfo = keyStatus.getKeyInfo();
-      // Reverting back the full path to key name
-      // Eg: a/b/c/d/e/file1 -> file1
-      omKeyInfo.setKeyName(OzoneFSUtils.getFileName(key));
       final long volumeId = omMetadataManager.getVolumeId(volume);
       final long bucketId = omMetadataManager.getBucketId(volume, bucket);
       final String dbKey = omMetadataManager.getOzonePathKey(volumeId, bucketId,
               omKeyInfo.getParentObjectID(), omKeyInfo.getFileName());
       boolean isDirectory = keyStatus.isDirectory();
-      operationResult = apply(omKeyInfo, trxnLogIndex);
-      omKeyInfo = omKeyInfo.toBuilder()
-          .setUpdateID(trxnLogIndex)
-          .build();
 
+      // Reverting back the full path to key name
+      // Eg: a/b/c/d/e/file1 -> file1
+      OmKeyInfo.Builder builder = omKeyInfo.toBuilder()
+          .setKeyName(OzoneFSUtils.getFileName(key));
+      operationResult = apply(builder, trxnLogIndex);
 
       // Update the modification time when updating ACLs of Key.
       long modificationTime = omKeyInfo.getModificationTime();
@@ -124,7 +122,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         modificationTime =
             getOmRequest().getRemoveAclRequest().getModificationTime();
       }
-      omKeyInfo.setModificationTime(modificationTime);
+
+      omKeyInfo = builder
+          .setModificationTime(modificationTime)
+          .setUpdateID(trxnLogIndex)
+          .build();
 
       // update cache.
       if (isDirectory) {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAddAclRequest.java

@@ -140,9 +140,9 @@ void onComplete(Result result, boolean operationResult,
   }
 
   @Override
-  boolean apply(OmKeyInfo omKeyInfo, long trxnLogIndex) {
+  boolean apply(OmKeyInfo.Builder builder, long trxnLogIndex) {
     // No need to check not null here, this will be never called with null.
-    return omKeyInfo.addAcl(ozoneAcls.get(0));
+    return builder.acls().add(ozoneAcls.get(0));
   }
 
   @Override

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAddAclRequestWithFSO.java

@@ -126,9 +126,9 @@ public OMKeyAddAclRequestWithFSO(
             getOmRequest().getUserInfo()));
   }
 
-  @Override boolean apply(OmKeyInfo omKeyInfo, long trxnLogIndex) {
+  @Override boolean apply(OmKeyInfo.Builder builder, long trxnLogIndex) {
     // No need to check not null here, this will be never called with null.
-    return omKeyInfo.addAcl(ozoneAcls.get(0));
+    return builder.acls().add(ozoneAcls.get(0));
   }
 
   @Override

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyRemoveAclRequest.java

@@ -141,9 +141,9 @@ void onComplete(Result result, boolean operationResult,
   }
 
   @Override
-  boolean apply(OmKeyInfo omKeyInfo, long trxnLogIndex) {
+  boolean apply(OmKeyInfo.Builder builder, long trxnLogIndex) {
     // No need to check not null here, this will be never called with null.
-    return omKeyInfo.removeAcl(ozoneAcls.get(0));
+    return builder.acls().remove(ozoneAcls.get(0));
   }
 
   @Override

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyRemoveAclRequestWithFSO.java

@@ -136,9 +136,9 @@ void onComplete(Result result, boolean operationResult,
   }
 
   @Override
-  boolean apply(OmKeyInfo omKeyInfo, long trxnLogIndex) {
+  boolean apply(OmKeyInfo.Builder builder, long trxnLogIndex) {
     // No need to check not null here, this will be never called with null.
-    return omKeyInfo.removeAcl(ozoneAcls.get(0));
+    return builder.acls().remove(ozoneAcls.get(0));
   }
 
   @Override

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeySetAclRequest.java

@@ -137,9 +137,9 @@ void onComplete(Result result, boolean operationResult,
   }
 
   @Override
-  boolean apply(OmKeyInfo omKeyInfo, long trxnLogIndex) {
+  boolean apply(OmKeyInfo.Builder builder, long trxnLogIndex) {
     // No need to check not null here, this will be never called with null.
-    return omKeyInfo.setAcls(ozoneAcls);
+    return builder.acls().set(ozoneAcls);
   }
 
   @Override

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeySetAclRequestWithFSO.java

@@ -129,9 +129,9 @@ void onComplete(Result result, boolean operationResult,
   }
 
   @Override
-  boolean apply(OmKeyInfo omKeyInfo, long trxnLogIndex) {
+  boolean apply(OmKeyInfo.Builder builder, long trxnLogIndex) {
     // No need to check not null here, this will be never called with null.
-    return omKeyInfo.setAcls(ozoneAcls);
+    return builder.acls().set(ozoneAcls);
   }
 
   @Override

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

@@ -397,7 +397,9 @@ private Optional<PurgePathRequest> prepareDeleteDirRequest(
         keyManager.getPendingDeletionSubDirs(volumeBucketId.getVolumeId(), volumeBucketId.getBucketId(),
             pendingDeletedDirInfo, keyInfo -> true, remainingNum);
     List<OmKeyInfo> subDirs = subDirDeleteResult.getKeysToDelete();
-    subDirs.forEach(omKeyInfo -> omKeyInfo.setAcls(Collections.emptyList()));
+    subDirs = subDirs.stream()
+        .map(omKeyInfo -> omKeyInfo.toBuilder().setAcls(Collections.emptyList()).build())
+        .collect(Collectors.toList());
     remainNum.addAndGet(-subDirs.size());
 
     OMMetadataManager omMetadataManager = keyManager.getMetadataManager();
@@ -417,7 +419,9 @@ private Optional<PurgePathRequest> prepareDeleteDirRequest(
         keyManager.getPendingDeletionSubFiles(volumeBucketId.getVolumeId(), volumeBucketId.getBucketId(),
             pendingDeletedDirInfo, keyInfo -> purgeDir || reclaimableFileFilter.apply(keyInfo), remainingNum);
     List<OmKeyInfo> subFiles = subFileDeleteResult.getKeysToDelete();
-    subFiles.forEach(omKeyInfo -> omKeyInfo.setAcls(Collections.emptyList()));
+    subFiles = subFiles.stream()
+        .map(omKeyInfo -> omKeyInfo.toBuilder().setAcls(Collections.emptyList()).build())
+        .collect(Collectors.toList());
     remainNum.addAndGet(-subFiles.size());
 
     if (LOG.isDebugEnabled()) {

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyCommitRequest.java

@@ -237,13 +237,13 @@ public void testAtomicRewrite() throws Exception {
             .map(OmKeyLocationInfo::getFromProtobuf)
             .collect(Collectors.toList());
 
+    List<OzoneAcl> acls = Collections.singletonList(OzoneAcl.parseAcl("user:foo:rw"));
     OmKeyInfo.Builder omKeyInfoBuilder = OMRequestTestUtils.createOmKeyInfo(
         volumeName, bucketName, keyName, replicationConfig, new OmKeyLocationInfoGroup(version, new ArrayList<>()));
-    omKeyInfoBuilder.setExpectedDataGeneration(1L);
-    OmKeyInfo omKeyInfo = omKeyInfoBuilder.build();
+    OmKeyInfo omKeyInfo = omKeyInfoBuilder.setExpectedDataGeneration(1L)
+        .addAcl(acls.get(0))
+        .build();
     omKeyInfo.appendNewBlocks(allocatedLocationList, false);
-    List<OzoneAcl> acls = Collections.singletonList(OzoneAcl.parseAcl("user:foo:rw"));
-    omKeyInfo.addAcl(acls.get(0));
 
     String openKey = addKeyToOpenKeyTable(allocatedLocationList, omKeyInfo);
     OmKeyInfo openKeyInfo = openKeyTable.get(openKey);

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/security/acl/OzoneNativeAclTestUtil.java

@@ -76,9 +76,10 @@ public static void addKeyAcl(
   ) throws IOException {
     final String objKey = metadataManager.getOzoneKey(volume, bucket, key);
     final Table<String, OmKeyInfo> keyTable = metadataManager.getKeyTable(bucketLayout);
-    final OmKeyInfo omKeyInfo = keyTable.get(objKey);
-
-    omKeyInfo.addAcl(ozoneAcl);
+    final OmKeyInfo omKeyInfo = keyTable.get(objKey)
+        .toBuilder()
+        .addAcl(ozoneAcl)
+        .build();
 
     keyTable.addCacheEntry(
         new CacheKey<>(objKey),
@@ -127,9 +128,10 @@ public static void setKeyAcl(
       List<OzoneAcl> ozoneAcls) throws IOException {
     final String objKey = metadataManager.getOzoneKey(volume, bucket, key);
     final Table<String, OmKeyInfo> keyTable = metadataManager.getKeyTable(bucketLayout);
-    final OmKeyInfo omKeyInfo = keyTable.get(objKey);
-
-    omKeyInfo.setAcls(ozoneAcls);
+    final OmKeyInfo omKeyInfo = keyTable.get(objKey)
+        .toBuilder()
+        .setAcls(ozoneAcls)
+        .build();
 
     keyTable.addCacheEntry(
         new CacheKey<>(objKey),