HDDS-10527. Rewrite key atomically

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -343,6 +343,7 @@ private OzoneConsts() {
   public static final String BUCKET_LAYOUT = "bucketLayout";
   public static final String TENANT = "tenant";
   public static final String USER_PREFIX = "userPrefix";
+  public static final String OVERWRITE_UPDATE_ID = "overwriteUpdateID";
 
   // For multi-tenancy
   public static final String TENANT_ID_USERNAME_DELIMITER = "$";

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java

@@ -469,6 +469,30 @@ public OzoneOutputStream createKey(String key, long size,
         .createKey(volumeName, name, key, size, replicationConfig, keyMetadata);
   }
 
+  /**
+   * Overwrite an existing key using optimistic locking. The existingKey must exist in Ozone to allow
+   * the new key to be created with the same name. Additionally, the existing Key must not have been
+   * modified since the time it's details were read. This is controlled by the updateID
+   * field in the existing Key. If the key is replaced or updated the updateID will change. If the
+   * updateID has changed since the existing Key was read, either the initial key create will fail,
+   * or the key will fail to commit after the data has been written as the checks are carried out
+   * both a key open and commit time.
+   *
+   * For now this feature only works on Object Store Buckets. FSO support will be added a later.
+   *
+   * @param existingKey       Name of the key to be created.
+   * @param replicationConfig Replication configuration.
+   * @return OzoneOutputStream to which the data has to be written.
+   * @throws IOException
+   */
+  public OzoneOutputStream overwriteKey(OzoneKeyDetails existingKey, ReplicationConfig replicationConfig)
+      throws IOException {
+    if (this.bucketLayout != BucketLayout.OBJECT_STORE) {
+      throw new IllegalArgumentException("Optimistic locking is only supported on Object Store Buckets");
+    }
+    return proxy.overwriteKey(existingKey, replicationConfig);
+  }
+
   /**
    * Creates a new key in the bucket, with default replication type RATIS and
    * with replication factor THREE.
@@ -1779,7 +1803,7 @@ private void addKeyPrefixInfoToResultList(String keyPrefix,
             keyInfo.getDataSize(), keyInfo.getCreationTime(),
             keyInfo.getModificationTime(),
             keyInfo.getReplicationConfig(),
-            keyInfo.isFile());
+            keyInfo.isFile(), keyInfo.getUpdateID());
         keysResultList.add(ozoneKey);
       }
     }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKey.java

@@ -70,11 +70,19 @@ public class OzoneKey {
    * Constructs OzoneKey from OmKeyInfo.
    *
    */
+
+  /**
+   * The update ID of an existing key. This will be null if this OzoneKey
+   * object has not been created from an existing key read from OM, as OM allocates
+   * the update ID on commit of the key.
+   */
+  private final Long updateID;
+
   @SuppressWarnings("parameternumber")
   public OzoneKey(String volumeName, String bucketName,
       String keyName, long size, long creationTime,
       long modificationTime, ReplicationConfig replicationConfig,
-      boolean isFile) {
+      boolean isFile, Long updateID) {
     this.volumeName = volumeName;
     this.bucketName = bucketName;
     this.name = keyName;
@@ -83,6 +91,16 @@ public OzoneKey(String volumeName, String bucketName,
     this.modificationTime = Instant.ofEpochMilli(modificationTime);
     this.replicationConfig = replicationConfig;
     this.isFile = isFile;
+    this.updateID = updateID;
+  }
+
+  @SuppressWarnings("parameternumber")
+  public OzoneKey(String volumeName, String bucketName,
+                  String keyName, long size, long creationTime,
+                  long modificationTime, ReplicationConfig replicationConfig,
+                  boolean isFile) {
+    this(volumeName, bucketName, keyName, size, creationTime, modificationTime, replicationConfig,
+        isFile, null);
   }
 
   @SuppressWarnings("parameternumber")
@@ -91,10 +109,19 @@ public OzoneKey(String volumeName, String bucketName,
                   long modificationTime, ReplicationConfig replicationConfig,
                   Map<String, String> metadata, boolean isFile) {
     this(volumeName, bucketName, keyName, size, creationTime,
-        modificationTime, replicationConfig, isFile);
+        modificationTime, replicationConfig, isFile, null);
     this.metadata.putAll(metadata);
   }
 
+  @SuppressWarnings("parameternumber")
+  public OzoneKey(String volumeName, String bucketName,
+                  String keyName, long size, long creationTime,
+                  long modificationTime, ReplicationConfig replicationConfig,
+                  Map<String, String> metadata, boolean isFile, Long updateID) {
+    this(volumeName, bucketName, keyName, size, creationTime,
+        modificationTime, replicationConfig, isFile, updateID);
+    this.metadata.putAll(metadata);
+  }
   /**
    * Returns Volume Name associated with the Key.
    *
@@ -179,6 +206,11 @@ public ReplicationConfig getReplicationConfig() {
     return replicationConfig;
   }
 
+  @JsonIgnore
+  public Long getUpdateID() {
+    return updateID;
+  }
+
   /**
    * Returns indicator if key is a file.
    * @return file
@@ -191,7 +223,7 @@ public static OzoneKey fromKeyInfo(OmKeyInfo keyInfo) {
     return new OzoneKey(keyInfo.getVolumeName(), keyInfo.getBucketName(),
         keyInfo.getKeyName(), keyInfo.getDataSize(), keyInfo.getCreationTime(),
         keyInfo.getModificationTime(), keyInfo.getReplicationConfig(),
-        keyInfo.getMetadata(), keyInfo.isFile());
+        keyInfo.getMetadata(), keyInfo.isFile(), keyInfo.getUpdateID());
   }
 
 }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKeyDetails.java

@@ -53,14 +53,31 @@ public OzoneKeyDetails(String volumeName, String bucketName, String keyName,
       Map<String, String> metadata,
       FileEncryptionInfo feInfo,
       CheckedSupplier<OzoneInputStream, IOException> contentSupplier,
-      boolean isFile) {
+      boolean isFile, Long updateID) {
     super(volumeName, bucketName, keyName, size, creationTime,
-        modificationTime, replicationConfig, metadata, isFile);
+        modificationTime, replicationConfig, metadata, isFile, updateID);
     this.ozoneKeyLocations = ozoneKeyLocations;
     this.feInfo = feInfo;
     this.contentSupplier = contentSupplier;
   }
 
+  /**
+   * Constructs OzoneKeyDetails from OmKeyInfo.
+   */
+  @SuppressWarnings("parameternumber")
+  public OzoneKeyDetails(String volumeName, String bucketName, String keyName,
+                         long size, long creationTime, long modificationTime,
+                         List<OzoneKeyLocation> ozoneKeyLocations,
+                         ReplicationConfig replicationConfig,
+                         Map<String, String> metadata,
+                         FileEncryptionInfo feInfo,
+                         CheckedSupplier<OzoneInputStream, IOException> contentSupplier,
+                         boolean isFile) {
+    this(volumeName, bucketName, keyName, size, creationTime,
+        modificationTime, ozoneKeyLocations, replicationConfig, metadata, feInfo, contentSupplier,
+        isFile, null);
+  }
+
   /**
    * Returns the location detail information of the specific Key.
    */

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -353,6 +353,21 @@ OzoneOutputStream createKey(String volumeName, String bucketName,
       Map<String, String> metadata)
       throws IOException;
 
+  /**
+   * Overwrite an existing key using optimistic locking. The OzoneKeyDetails passed must contain
+   * the updateID of the key to be overwritten. The existing key must also exist on
+   * OM and have the same UpdateID as the one passed in or the key create / commit will fail.
+   *
+   * Currently does not work for FSO buckets. Support will be added later.
+   *
+   * @param keyToOverwrite Existing key to overwrite
+   * @param replicationConfig The replication configuration for the new key
+   * @return {@link OzoneOutputStream}
+   * @throws IOException
+   */
+  OzoneOutputStream overwriteKey(OzoneKeyDetails keyToOverwrite, ReplicationConfig replicationConfig)
+      throws IOException;
+
   /**
    * Writes a key in an existing bucket.
    * @param volumeName Name of the Volume

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

@@ -1374,26 +1374,7 @@ public OzoneOutputStream createKey(
       ReplicationConfig replicationConfig,
       Map<String, String> metadata)
       throws IOException {
-    verifyVolumeName(volumeName);
-    verifyBucketName(bucketName);
-    if (checkKeyNameEnabled) {
-      HddsClientUtils.verifyKeyName(keyName);
-    }
-    HddsClientUtils.checkNotNull(keyName);
-    if (omVersion
-        .compareTo(OzoneManagerVersion.ERASURE_CODED_STORAGE_SUPPORT) < 0) {
-      if (replicationConfig != null &&
-          replicationConfig.getReplicationType()
-              == HddsProtos.ReplicationType.EC) {
-        throw new IOException("Can not set the replication of the key to"
-            + " Erasure Coded replication, as OzoneManager does not support"
-            + " Erasure Coded replication.");
-      }
-    }
-
-    if (replicationConfig != null) {
-      replicationConfigValidator.validate(replicationConfig);
-    }
+    createKeyPreChecks(volumeName, bucketName, keyName, replicationConfig);
 
     OmKeyArgs.Builder builder = new OmKeyArgs.Builder()
         .setVolumeName(volumeName)
@@ -1416,6 +1397,63 @@ public OzoneOutputStream createKey(
     return createOutputStream(openKey);
   }
 
+  @Override
+  public OzoneOutputStream overwriteKey(OzoneKeyDetails keyToOverwrite, ReplicationConfig replicationConfig)
+      throws IOException {
+    if (keyToOverwrite == null) {
+      throw new IllegalArgumentException("KeyToOverwrite cannot be null");
+    }
+    if (keyToOverwrite.getUpdateID() == null) {
+      throw new IllegalArgumentException("KeyToOverwrite UpdateID cannot be null");
+    }
+    createKeyPreChecks(keyToOverwrite.getVolumeName(), keyToOverwrite.getBucketName(),
+        keyToOverwrite.getName(), replicationConfig);
+
+    OmKeyArgs.Builder builder = new OmKeyArgs.Builder()
+        .setVolumeName(keyToOverwrite.getVolumeName())
+        .setBucketName(keyToOverwrite.getBucketName())
+        .setKeyName(keyToOverwrite.getName())
+        .setDataSize(keyToOverwrite.getDataSize())
+        .setReplicationConfig(replicationConfig)
+        .addAllMetadataGdpr(keyToOverwrite.getMetadata())
+        .setLatestVersionLocation(getLatestVersionLocation)
+        .setOverwriteUpdateID(keyToOverwrite.getUpdateID());
+
+    OpenKeySession openKey = ozoneManagerClient.openKey(builder.build());
+    // For bucket with layout OBJECT_STORE, when create an empty file (size=0),
+    // OM will set DataSize to OzoneConfigKeys#OZONE_SCM_BLOCK_SIZE,
+    // which will cause S3G's atomic write length check to fail,
+    // so reset size to 0 here.
+    if (isS3GRequest.get() && keyToOverwrite.getDataSize() == 0) {
+      openKey.getKeyInfo().setDataSize(0);
+    }
+    return createOutputStream(openKey);
+  }
+
+  private void createKeyPreChecks(String volumeName, String bucketName, String keyName,
+      ReplicationConfig replicationConfig) throws IOException {
+    verifyVolumeName(volumeName);
+    verifyBucketName(bucketName);
+    if (checkKeyNameEnabled) {
+      HddsClientUtils.verifyKeyName(keyName);
+    }
+    HddsClientUtils.checkNotNull(keyName);
+    if (omVersion
+        .compareTo(OzoneManagerVersion.ERASURE_CODED_STORAGE_SUPPORT) < 0) {
+      if (replicationConfig != null &&
+          replicationConfig.getReplicationType()
+              == HddsProtos.ReplicationType.EC) {
+        throw new IOException("Can not set the replication of the key to"
+            + " Erasure Coded replication, as OzoneManager does not support"
+            + " Erasure Coded replication.");
+      }
+    }
+
+    if (replicationConfig != null) {
+      replicationConfigValidator.validate(replicationConfig);
+    }
+  }
+
   @Override
   public OzoneDataStreamOutput createStreamKey(
       String volumeName, String bucketName, String keyName, long size,
@@ -1652,7 +1690,8 @@ public List<OzoneKey> listKeys(String volumeName, String bucketName,
               key.getCreationTime(),
               key.getModificationTime(),
               key.getReplicationConfig(),
-              key.isFile()))
+              key.isFile(),
+              key.getUpdateID()))
           .collect(Collectors.toList());
     }
   }
@@ -1703,7 +1742,7 @@ private OzoneKeyDetails getOzoneKeyDetails(OmKeyInfo keyInfo) {
         keyInfo.getModificationTime(), ozoneKeyLocations,
         keyInfo.getReplicationConfig(), keyInfo.getMetadata(),
         keyInfo.getFileEncryptionInfo(),
-        () -> getInputStreamWithRetryFunction(keyInfo), keyInfo.isFile());
+        () -> getInputStreamWithRetryFunction(keyInfo), keyInfo.isFile(), keyInfo.getUpdateID());
   }
 
   @Override

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

@@ -52,6 +52,12 @@ public final class OmKeyArgs implements Auditable {
   private final boolean recursive;
   private final boolean headOp;
   private final boolean forceUpdateContainerCacheFromSCM;
+  // Overwrite updateID when used in key creation indicates that a
+  // key with the same keyName should exist with the given update ID.
+  // Upon commit, the updateID of that existing key should be unchanged.
+  // This is a form of optimistic locking, to ensure the key is not modified between
+  // the time the key is read and the time the key is written.
+  private Long overwriteUpdateID = null;
 
   private OmKeyArgs(Builder b) {
     this.volumeName = b.volumeName;
@@ -70,6 +76,7 @@ private OmKeyArgs(Builder b) {
     this.recursive = b.recursive;
     this.headOp = b.headOp;
     this.forceUpdateContainerCacheFromSCM = b.forceUpdateContainerCacheFromSCM;
+    this.overwriteUpdateID = b.overwriteUpdateID;
   }
 
   public boolean getIsMultipartKey() {
@@ -144,6 +151,10 @@ public boolean isForceUpdateContainerCacheFromSCM() {
     return forceUpdateContainerCacheFromSCM;
   }
 
+  public Long getOverwriteUpdateID() {
+    return overwriteUpdateID;
+  }
+
   @Override
   public Map<String, String> toAuditMap() {
     Map<String, String> auditMap = new LinkedHashMap<>();
@@ -166,7 +177,7 @@ public void addLocationInfo(OmKeyLocationInfo locationInfo) {
   }
 
   public OmKeyArgs.Builder toBuilder() {
-    return new OmKeyArgs.Builder()
+    OmKeyArgs.Builder builder = new OmKeyArgs.Builder()
         .setVolumeName(volumeName)
         .setBucketName(bucketName)
         .setKeyName(keyName)
@@ -182,11 +193,16 @@ public OmKeyArgs.Builder toBuilder() {
         .setLatestVersionLocation(latestVersionLocation)
         .setAcls(acls)
         .setForceUpdateContainerCacheFromSCM(forceUpdateContainerCacheFromSCM);
+
+    if (overwriteUpdateID != null) {
+      builder.setOverwriteUpdateID(overwriteUpdateID);
+    }
+    return builder;
   }
 
   @Nonnull
   public KeyArgs toProtobuf() {
-    return KeyArgs.newBuilder()
+    KeyArgs.Builder builder = KeyArgs.newBuilder()
         .setVolumeName(getVolumeName())
         .setBucketName(getBucketName())
         .setKeyName(getKeyName())
@@ -195,8 +211,11 @@ public KeyArgs toProtobuf() {
         .setLatestVersionLocation(getLatestVersionLocation())
         .setHeadOp(isHeadOp())
         .setForceUpdateContainerCacheFromSCM(
-            isForceUpdateContainerCacheFromSCM())
-        .build();
+            isForceUpdateContainerCacheFromSCM());
+    if (overwriteUpdateID != null) {
+      builder.setOverwriteUpdateID(overwriteUpdateID);
+    }
+    return builder.build();
   }
 
   /**
@@ -219,6 +238,7 @@ public static class Builder {
     private boolean recursive;
     private boolean headOp;
     private boolean forceUpdateContainerCacheFromSCM;
+    private Long overwriteUpdateID = null;
 
     public Builder setVolumeName(String volume) {
       this.volumeName = volume;
@@ -313,6 +333,11 @@ public Builder setForceUpdateContainerCacheFromSCM(boolean value) {
       return this;
     }
 
+    public Builder setOverwriteUpdateID(long overwriteUpdateID) {
+      this.overwriteUpdateID = overwriteUpdateID;
+      return this;
+    }
+
     public OmKeyArgs build() {
       return new OmKeyArgs(this);
     }