HDDS-10527. Rewrite key atomically

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

@@ -343,6 +343,9 @@ 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_OBJECT_ID = "overwriteObjectID";
+  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,25 @@ 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 existingKey must not have been
+   * modified since the time its details were read. This is controlled by the objectID and updateID
+   * fields in the existingKey. If the key is replaced the objectID will change. If it has been updated
+   * (eg appended) the updateID will change. If either of these have changed since the existingKey
+   * was read, either the initial key create will fail, or the key will fail to commit after the data
+   * has been written.
+   *
+   * @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 {
+    return proxy.overwriteKey(existingKey, replicationConfig);
+  }
+
   /**
    * Creates a new key in the bucket, with default replication type RATIS and
    * with replication factor THREE.
@@ -1779,7 +1798,7 @@ private void addKeyPrefixInfoToResultList(String keyPrefix,
             keyInfo.getDataSize(), keyInfo.getCreationTime(),
             keyInfo.getModificationTime(),
             keyInfo.getReplicationConfig(),
-            keyInfo.isFile());
+            keyInfo.isFile(), keyInfo.getObjectID(), 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 object and update ID of an existing key. These will be null
+   * if the key is not yet created on OM and read from OM.
+   */
+  private final Long objectID;
+  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 objectID, Long updateID) {
     this.volumeName = volumeName;
     this.bucketName = bucketName;
     this.name = keyName;
@@ -83,6 +91,17 @@ public OzoneKey(String volumeName, String bucketName,
     this.modificationTime = Instant.ofEpochMilli(modificationTime);
     this.replicationConfig = replicationConfig;
     this.isFile = isFile;
+    this.objectID = objectID;
+    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, null);
   }
 
   @SuppressWarnings("parameternumber")
@@ -91,10 +110,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, 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 objectID, Long updateID) {
+    this(volumeName, bucketName, keyName, size, creationTime,
+        modificationTime, replicationConfig, isFile, objectID, updateID);
+    this.metadata.putAll(metadata);
+  }
   /**
    * Returns Volume Name associated with the Key.
    *
@@ -179,6 +207,16 @@ public ReplicationConfig getReplicationConfig() {
     return replicationConfig;
   }
 
+  @JsonIgnore
+  public Long getObjectID() {
+    return objectID;
+  }
+
+  @JsonIgnore
+  public Long getUpdateID() {
+    return updateID;
+  }
+
   /**
    * Returns indicator if key is a file.
    * @return file
@@ -191,7 +229,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.getObjectID(), 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 objectID, Long updateID) {
     super(volumeName, bucketName, keyName, size, creationTime,
-        modificationTime, replicationConfig, metadata, isFile);
+        modificationTime, replicationConfig, metadata, isFile, objectID, 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, 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,19 @@ 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 objectID and updateID of the key to be overwritten. The existing key must also exist on
+   * OM and have the same ObjectID and UpdateID as the one passed in or the key create / commit
+   * will fail.
+   * @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

@@ -1378,26 +1378,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)
@@ -1420,6 +1401,67 @@ 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.getObjectID() == null) {
+      throw new IllegalArgumentException("KeyToOverwrite ObjectID 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)
+        .setOverwriteObjectID(keyToOverwrite.getObjectID())
+        .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,
@@ -1643,7 +1685,9 @@ public List<OzoneKey> listKeys(String volumeName, String bucketName,
               key.getCreationTime(),
               key.getModificationTime(),
               key.getReplicationConfig(),
-              key.isFile()))
+              key.isFile(),
+              null,
+              null))
           .collect(Collectors.toList());
     } else {
       List<OmKeyInfo> keys = ozoneManagerClient.listKeys(
@@ -1656,7 +1700,9 @@ public List<OzoneKey> listKeys(String volumeName, String bucketName,
               key.getCreationTime(),
               key.getModificationTime(),
               key.getReplicationConfig(),
-              key.isFile()))
+              key.isFile(),
+              key.getObjectID(),
+              key.getUpdateID()))
           .collect(Collectors.toList());
     }
   }
@@ -1707,7 +1753,8 @@ private OzoneKeyDetails getOzoneKeyDetails(OmKeyInfo keyInfo) {
         keyInfo.getModificationTime(), ozoneKeyLocations,
         keyInfo.getReplicationConfig(), keyInfo.getMetadata(),
         keyInfo.getFileEncryptionInfo(),
-        () -> getInputStreamWithRetryFunction(keyInfo), keyInfo.isFile());
+        () -> getInputStreamWithRetryFunction(keyInfo), keyInfo.isFile(),
+        keyInfo.getObjectID(), keyInfo.getUpdateID());
   }
 
   @Override