HDDS-10388. Make WithMetadata immutable

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

@@ -1625,8 +1625,8 @@ public OzoneInputStream getKey(
             .setParentObjectID(keyInfo.getParentObjectID())
             .setFileChecksum(keyInfo.getFileChecksum())
             .setOwnerName(keyInfo.getOwnerName())
+            .addAllMetadata(keyInfo.getMetadata())
             .build();
-        dnKeyInfo.setMetadata(keyInfo.getMetadata());
         dnKeyInfo.setKeyLocationVersions(keyLocationInfoGroups);
 
         blocks.put(dn, createInputStream(dnKeyInfo, Function.identity()));

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

@@ -541,23 +541,29 @@ public static File createOMDir(String dirPath) {
    */
   public static RepeatedOmKeyInfo prepareKeyForDelete(long bucketId, OmKeyInfo keyInfo,
       long trxnLogIndex) {
+    OmKeyInfo sanitizedKeyInfo = keyInfo;
     // If this key is in a GDPR enforced bucket, then before moving
     // KeyInfo to deletedTable, remove the GDPR related metadata and
     // FileEncryptionInfo from KeyInfo.
     if (Boolean.parseBoolean(
             keyInfo.getMetadata().get(OzoneConsts.GDPR_FLAG))
     ) {
-      keyInfo.getMetadata().remove(OzoneConsts.GDPR_FLAG);
-      keyInfo.getMetadata().remove(OzoneConsts.GDPR_ALGORITHM);
-      keyInfo.getMetadata().remove(OzoneConsts.GDPR_SECRET);
-      keyInfo.clearFileEncryptionInfo();
+      sanitizedKeyInfo = sanitizedKeyInfo.withMetadataMutations(metadata -> {
+        metadata.remove(OzoneConsts.GDPR_FLAG);
+        metadata.remove(OzoneConsts.GDPR_ALGORITHM);
+        metadata.remove(OzoneConsts.GDPR_SECRET);
+      });
+      sanitizedKeyInfo.clearFileEncryptionInfo();
     }
 
     // Set the updateID
-    keyInfo.setUpdateID(trxnLogIndex);
+    sanitizedKeyInfo.setUpdateID(trxnLogIndex);
+    if (sanitizedKeyInfo != keyInfo) {
+      keyInfo.setUpdateID(trxnLogIndex);
+    }
 
     //The key doesn't exist in deletedTable, so create a new instance.
-    return new RepeatedOmKeyInfo(keyInfo, bucketId);
+    return new RepeatedOmKeyInfo(sanitizedKeyInfo, bucketId);
   }
 
   /**

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

@@ -25,6 +25,7 @@
 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;
 import org.apache.hadoop.fs.FileChecksum;
@@ -193,12 +194,38 @@ public String getOwnerName() {
     return ownerName;
   }
 
-  public void setCommittedKeyDeletedFlag(boolean val) {
+  public OmKeyInfo withCommittedKeyDeletedFlag(boolean val) {
     if (val) {
-      this.getMetadata().put(COMMITTED_KEY_DELETED_FLAG, "true");
-    } else {
-      this.getMetadata().remove(COMMITTED_KEY_DELETED_FLAG);
+      return withMetadataMutations(
+          metadata -> metadata.put(COMMITTED_KEY_DELETED_FLAG, "true"));
     }
+    return withMetadataMutations(
+        metadata -> metadata.remove(COMMITTED_KEY_DELETED_FLAG));
+  }
+
+  /**
+   * Returns a new {@link OmKeyInfo} instance with metadata updated by the
+   * provided mutator.
+   *
+   * @param metadataUpdater a function that applies mutations to a copy of the metadata
+   * @return a new {@link OmKeyInfo} instance with updated metadata
+   */
+  public OmKeyInfo withMetadataMutations(
+      Consumer<Map<String, String>> metadataUpdater) {
+    Objects.requireNonNull(metadataUpdater, "metadataUpdater == null");
+    Map<String, String> metadataCopy = new HashMap<>(getMetadata());
+    metadataUpdater.accept(metadataCopy);
+    return toBuilder().setMetadata(metadataCopy).build();
+  }
+
+  /**
+   * Returns a new {@link OmKeyInfo} with metadata replaced by the provided
+   * map.
+   * @param metadata the metadata to set
+   * @return a new {@link OmKeyInfo}
+   */
+  public OmKeyInfo withMetadata(Map<String, String> metadata) {
+    return toBuilder().setMetadata(metadata).build();
   }
 
   public boolean isDeletedKeyCommitted() {
@@ -499,6 +526,28 @@ public Builder() {
 
     public Builder(OmKeyInfo obj) {
       super(obj);
+      this.volumeName = obj.volumeName;
+      this.bucketName = obj.bucketName;
+      this.keyName = obj.keyName;
+      this.ownerName = obj.ownerName;
+      this.dataSize = obj.dataSize;
+      this.creationTime = obj.creationTime;
+      this.modificationTime = obj.modificationTime;
+      this.replicationConfig = obj.replicationConfig;
+      this.encInfo = obj.encInfo;
+      this.fileName = obj.fileName;
+      this.fileChecksum = obj.fileChecksum;
+      this.isFile = obj.isFile;
+      this.expectedDataGeneration = obj.expectedDataGeneration;
+      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(),
+                  keyLocationVersion.getLocationList(),
+                  keyLocationVersion.isMultipartKey())));
     }
 
     public Builder setVolumeName(String volume) {
@@ -569,6 +618,12 @@ public Builder addAllMetadata(Map<String, String> newMetadata) {
       return this;
     }
 
+    @Override
+    public Builder setMetadata(Map<String, String> map) {
+      super.setMetadata(map);
+      return this;
+    }
+
     public Builder setFileEncryptionInfo(FileEncryptionInfo feInfo) {
       this.encInfo = feInfo;
       return this;
@@ -882,44 +937,13 @@ public int hashCode() {
   /**
    * Return a new copy of the object.
    */
+  public Builder toBuilder() {
+    return new Builder(this);
+  }
+
   @Override
   public OmKeyInfo copyObject() {
-    OmKeyInfo.Builder builder = new OmKeyInfo.Builder(this)
-        .setVolumeName(volumeName)
-        .setBucketName(bucketName)
-        .setKeyName(keyName)
-        .setOwnerName(ownerName)
-        .setCreationTime(creationTime)
-        .setModificationTime(modificationTime)
-        .setDataSize(dataSize)
-        .setReplicationConfig(replicationConfig)
-        .setFileEncryptionInfo(encInfo)
-        .setAcls(acls)
-        .setFileName(fileName)
-        .setFile(isFile);
-
-    keyLocationVersions.forEach(keyLocationVersion ->
-        builder.addOmKeyLocationInfoGroup(
-            new OmKeyLocationInfoGroup(keyLocationVersion.getVersion(),
-                keyLocationVersion.getLocationList(),
-                keyLocationVersion.isMultipartKey())));
-
-    if (getMetadata() != null) {
-      getMetadata().forEach(builder::addMetadata);
-    }
-
-    if (getTags() != null) {
-      getTags().forEach(builder::addTag);
-    }
-
-    if (fileChecksum != null) {
-      builder.setFileChecksum(fileChecksum);
-    }
-    if (expectedDataGeneration != null) {
-      builder.setExpectedDataGeneration(expectedDataGeneration);
-    }
-
-    return builder.build();
+    return new Builder(this).build();
   }
 
   /**

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

@@ -17,26 +17,31 @@
 
 package org.apache.hadoop.ozone.om.helpers;
 
+import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import net.jcip.annotations.Immutable;
 
 /**
  * Mixin class to handle custom metadata.
  */
+@Immutable
 public abstract class WithMetadata {
 
-  private Map<String, String> metadata;
+  private final Map<String, String> metadata;
 
   protected WithMetadata() {
-    metadata = new ConcurrentHashMap<>();
+    metadata = ImmutableMap.of();
   }
 
   protected WithMetadata(Builder b) {
-    metadata = b.metadata;
+    metadata = b.metadata == null ? ImmutableMap.of()
+        : ImmutableMap.copyOf(b.metadata);
   }
 
   protected WithMetadata(WithMetadata other) {
-    metadata = new ConcurrentHashMap<>(other.getMetadata());
+    metadata = other.getMetadata() == null ? ImmutableMap.of()
+        : ImmutableMap.copyOf(other.getMetadata());
   }
 
   /**
@@ -46,13 +51,6 @@ public final Map<String, String> getMetadata() {
     return metadata;
   }
 
-  /**
-   * Set custom key value metadata.
-   */
-  public final void setMetadata(Map<String, String> metadata) {
-    this.metadata = metadata;
-  }
-
   /** Builder for {@link WithMetadata}. */
   public static class Builder {
     private final Map<String, String> metadata;

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

@@ -64,7 +64,8 @@ public void protobufConversion() throws IOException {
     assertEquals(key, keyAfterSerialization);
 
     assertFalse(key.isHsync());
-    key.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID, "clientid");
+    key = key.withMetadataMutations(
+        metadata -> metadata.put(OzoneConsts.HSYNC_CLIENT_ID, "clientid"));
     assertTrue(key.isHsync());
     assertEquals(5678L, key.getExpectedDataGeneration());
   }

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

@@ -4454,7 +4454,8 @@ public void testKeyReadWriteForGDPR() throws Exception {
     OmKeyInfo omKeyInfo = omMetadataManager.getKeyTable(BucketLayout.OBJECT_STORE)
         .get(omMetadataManager.getOzoneKey(volumeName, bucketName, keyName));
 
-    omKeyInfo.getMetadata().remove(OzoneConsts.GDPR_FLAG);
+    omKeyInfo = omKeyInfo.withMetadataMutations(
+        metadata -> metadata.remove(OzoneConsts.GDPR_FLAG));
 
     omMetadataManager.getKeyTable(BucketLayout.OBJECT_STORE)
         .put(omMetadataManager.getOzoneKey(volumeName, bucketName, keyName),

hadoop-ozone/interface-storage/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmPrefixInfo.java

@@ -131,7 +131,9 @@ public void testGetProtobuf() {
         IAccessAuthorizer.ACLIdentityType.USER,
         username, IAccessAuthorizer.ACLType.WRITE,
         ACCESS);
-    omPrefixInfo.getMetadata().put("key", "value");
+    omPrefixInfo = new OmPrefixInfo.Builder(omPrefixInfo)
+        .addMetadata("key", "value")
+        .build();
     OzoneManagerStorageProtos.PersistedPrefixInfo pi =
         omPrefixInfo.getProtobuf();
     assertEquals(testPath, pi.getName());

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMRecoverLeaseRequest.java

@@ -228,7 +228,9 @@ private RecoverLeaseResponse doWork(OzoneManager ozoneManager,
         throw new OMException("Open Key " + keyName + " updated recently and is inside soft limit period",
             KEY_UNDER_LEASE_SOFT_LIMIT_PERIOD);
       }
-      openKeyInfo.getMetadata().put(OzoneConsts.LEASE_RECOVERY, "true");
+      openKeyInfo = openKeyInfo.toBuilder()
+          .addMetadata(OzoneConsts.LEASE_RECOVERY, "true")
+          .build();
       openKeyInfo.setUpdateID(transactionLogIndex);
       openKeyInfo.setModificationTime(Time.now());
       // add to cache.

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

@@ -173,7 +173,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
                 parentId, processed.keyInfo.getFileName(), hsyncClientId);
             OmKeyInfo openKeyInfo = omMetadataManager.getOpenKeyTable(getBucketLayout()).get(dbOpenKey);
             if (openKeyInfo != null) {
-              openKeyInfo.getMetadata().put(DELETED_HSYNC_KEY, "true");
+              openKeyInfo = openKeyInfo.withMetadataMutations(
+                  metadata -> metadata.put(DELETED_HSYNC_KEY, "true"));
               openKeyInfoMap.put(dbOpenKey, openKeyInfo);
             }
           }

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

@@ -274,7 +274,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         dbOpenKeyToDeleteKey = omMetadataManager.getOpenKey(volumeName, bucketName,
             keyName, Long.parseLong(keyToDelete.getMetadata().get(OzoneConsts.HSYNC_CLIENT_ID)));
         openKeyToDelete = omMetadataManager.getOpenKeyTable(getBucketLayout()).get(dbOpenKeyToDeleteKey);
-        openKeyToDelete.getMetadata().put(OzoneConsts.OVERWRITTEN_HSYNC_KEY, "true");
+        openKeyToDelete = openKeyToDelete.toBuilder()
+            .addMetadata(OzoneConsts.OVERWRITTEN_HSYNC_KEY, "true")
+            .build();
         openKeyToDelete.setModificationTime(Time.now());
         openKeyToDelete.setUpdateID(trxnLogIndex);
         omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
@@ -288,7 +290,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       if (isHSync) {
         if (!OmKeyHSyncUtil.isHSyncedPreviously(omKeyInfo, clientIdString, dbOpenKey)) {
           // Update open key as well if it is the first hsync of this key
-          omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID, clientIdString);
+          omKeyInfo = omKeyInfo.withMetadataMutations(
+              metadata -> metadata.put(OzoneConsts.HSYNC_CLIENT_ID, clientIdString));
           newOpenKeyInfo = omKeyInfo.copyObject();
         }
       }
@@ -298,8 +301,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       // not persisted in the key table.
       omKeyInfo.setExpectedDataGeneration(null);
 
-      omKeyInfo.getMetadata().putAll(KeyValueUtil.getFromProtobuf(
-          commitKeyArgs.getMetadataList()));
+      omKeyInfo = omKeyInfo.withMetadataMutations(metadata ->
+          metadata.putAll(KeyValueUtil.getFromProtobuf(
+              commitKeyArgs.getMetadataList())));
       omKeyInfo.setDataSize(commitKeyArgs.getDataSize());
 
       // Update the block length for each block, return the allocated but
@@ -343,9 +347,12 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         long totalNamespace = 0;
         if (!oldVerKeyInfo.getOmKeyInfoList().isEmpty()) {
           oldKeyVersionsToDeleteMap.put(delKeyName, oldVerKeyInfo);
-          for (OmKeyInfo olderKeyVersions : oldVerKeyInfo.getOmKeyInfoList()) {
-            olderKeyVersions.setCommittedKeyDeletedFlag(true);
-            totalSize += sumBlockLengths(olderKeyVersions);
+          List<OmKeyInfo> oldKeys = oldVerKeyInfo.getOmKeyInfoList();
+          for (int i = 0; i < oldKeys.size(); i++) {
+            OmKeyInfo updatedOlderKeyVersions =
+                oldKeys.get(i).withCommittedKeyDeletedFlag(true);
+            oldKeys.set(i, updatedOlderKeyVersions);
+            totalSize += sumBlockLengths(updatedOlderKeyVersions);
             totalNamespace += 1;
           }
         }
@@ -379,10 +386,12 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
             dbOpenKey, trxnLogIndex);
 
         // Prevent hsync metadata from getting committed to the final key
-        omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
-        if (isRecovery) {
-          omKeyInfo.getMetadata().remove(OzoneConsts.LEASE_RECOVERY);
-        }
+        omKeyInfo = omKeyInfo.withMetadataMutations(metadata -> {
+          metadata.remove(OzoneConsts.HSYNC_CLIENT_ID);
+          if (isRecovery) {
+            metadata.remove(OzoneConsts.LEASE_RECOVERY);
+          }
+        });
       } else if (newOpenKeyInfo != null) {
         // isHSync is true and newOpenKeyInfo is set, update OpenKeyTable
         omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(