HDDS-10527. Rewrite key atomically

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.
@@ -1287,7 +1306,7 @@ protected List<OzoneKey> buildKeysWithKeyPrefix(
                 keyInfo.getBucketName(), keyName,
                 keyInfo.getDataSize(), keyInfo.getCreationTime(),
                 keyInfo.getModificationTime(),
-                keyInfo.getReplicationConfig(), keyInfo.isFile());
+                keyInfo.getReplicationConfig(), keyInfo.isFile(), null, null);
           })
           .filter(key -> StringUtils.startsWith(key.getName(), getKeyPrefix()))
           .collect(Collectors.toList());
@@ -1675,7 +1694,7 @@ private boolean getChildrenKeys(String keyPrefix, String startKey,
             keyInfo.getDataSize(), keyInfo.getCreationTime(),
             keyInfo.getModificationTime(),
             keyInfo.getReplicationConfig(),
-            keyInfo.isFile());
+            keyInfo.isFile(), null, null);
 
         keysResultList.add(ozoneKey);
 
@@ -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,15 +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,
-                  Map<String, String> metadata, boolean isFile) {
+                  Map<String, String> metadata, boolean isFile, Long objectID, Long updateID) {
     this(volumeName, bucketName, keyName, size, creationTime,
-        modificationTime, replicationConfig, isFile);
+        modificationTime, replicationConfig, isFile, objectID, updateID);
     this.metadata.putAll(metadata);
   }
 
@@ -179,6 +189,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 +211,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,9 +53,9 @@ 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;

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,70 @@ 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())
+        // TODO - if we are effectively cloning a key, probably the ACLs should
+        //        be copied over server side. I am not too sure how this works.
+        .setAcls(getAclList())
+        .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 +1688,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 +1703,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 +1756,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

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

@@ -52,6 +52,13 @@ public final class OmKeyArgs implements Auditable {
   private final boolean recursive;
   private final boolean headOp;
   private final boolean forceUpdateContainerCacheFromSCM;
+  // Overwrite objectID and updateID when used in key creation indicate that a
+  // key with the same keyName should exist with the given object and update IDs,
+  // and upon commit, the object and 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 overwriteObjectID = null;
+  private Long overwriteUpdateID = null;
 
   private OmKeyArgs(Builder b) {
     this.volumeName = b.volumeName;
@@ -70,6 +77,8 @@ private OmKeyArgs(Builder b) {
     this.recursive = b.recursive;
     this.headOp = b.headOp;
     this.forceUpdateContainerCacheFromSCM = b.forceUpdateContainerCacheFromSCM;
+    this.overwriteObjectID = b.overwriteObjectID;
+    this.overwriteUpdateID = b.overwriteUpdateID;
   }
 
   public boolean getIsMultipartKey() {
@@ -144,6 +153,14 @@ public boolean isForceUpdateContainerCacheFromSCM() {
     return forceUpdateContainerCacheFromSCM;
   }
 
+  public Long getOverwriteObjectID() {
+    return overwriteObjectID;
+  }
+
+  public Long getOverwriteUpdateID() {
+    return overwriteUpdateID;
+  }
+
   @Override
   public Map<String, String> toAuditMap() {
     Map<String, String> auditMap = new LinkedHashMap<>();
@@ -166,7 +183,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 +199,19 @@ public OmKeyArgs.Builder toBuilder() {
         .setLatestVersionLocation(latestVersionLocation)
         .setAcls(acls)
         .setForceUpdateContainerCacheFromSCM(forceUpdateContainerCacheFromSCM);
+
+    if (overwriteUpdateID != null) {
+      builder.setOverwriteObjectID(overwriteUpdateID);
+    }
+    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 +220,14 @@ public KeyArgs toProtobuf() {
         .setLatestVersionLocation(getLatestVersionLocation())
         .setHeadOp(isHeadOp())
         .setForceUpdateContainerCacheFromSCM(
-            isForceUpdateContainerCacheFromSCM())
-        .build();
+            isForceUpdateContainerCacheFromSCM());
+    if (overwriteUpdateID != null) {
+      builder.setOverwriteObjectID(overwriteUpdateID);
+    }
+    if (overwriteUpdateID != null) {
+      builder.setOverwriteUpdateID(overwriteUpdateID);
+    }
+    return builder.build();
   }
 
   /**
@@ -219,6 +250,8 @@ public static class Builder {
     private boolean recursive;
     private boolean headOp;
     private boolean forceUpdateContainerCacheFromSCM;
+    private Long overwriteObjectID = null;
+    private Long overwriteUpdateID = null;
 
     public Builder setVolumeName(String volume) {
       this.volumeName = volume;
@@ -313,6 +346,16 @@ public Builder setForceUpdateContainerCacheFromSCM(boolean value) {
       return this;
     }
 
+    public Builder setOverwriteObjectID(long overwriteObjectID) {
+      this.overwriteObjectID = overwriteObjectID;
+      return this;
+    }
+
+    public Builder setOverwriteUpdateID(long overwriteUpdateID) {
+      this.overwriteUpdateID = overwriteUpdateID;
+      return this;
+    }
+
     public OmKeyArgs build() {
       return new OmKeyArgs(this);
     }