HDDS-6171. Create an API to change Bucket Owner

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

@@ -905,6 +905,17 @@ public OzoneMultipartUploadList listMultipartUploads(String prefix)
     return proxy.listMultipartUploads(volumeName, getName(), prefix);
   }
 
+  /**
+   * Sets/Changes the owner of this Bucket.
+   * @param userName new owner
+   * @throws IOException
+   */
+  public boolean setOwner(String userName) throws IOException{
+    boolean result = proxy.setBucketOwner(volumeName, name, userName);
+    this.owner = userName;
+    return result;
+  }
+
   /**
    * An Iterator to iterate over {@link OzoneKey} list.
    */

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

@@ -772,4 +772,14 @@ OzoneKey headObject(String volumeName, String bucketName,
    * Clears the S3 Authentication information attached to the thread.
    */
   void clearTheadLocalS3Auth();
+
+  /**
+   * Sets the owner of bucket.
+   * @param volumeName Name of the Volume
+   * @param bucketName Name of the Bucket
+   * @param owner to be set for the bucket
+   * @throws IOException
+   */
+  boolean setBucketOwner(String volumeName, String bucketName,
+      String owner) throws IOException;
 }

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

@@ -1579,4 +1579,17 @@ public S3Auth getThreadLocalS3Auth() {
   public void clearTheadLocalS3Auth() {
     ozoneManagerClient.clearThreadLocalS3Auth();
   }
+
+  @Override
+  public boolean setBucketOwner(String volumeName, String bucketName,
+      String owner) throws IOException {
+    verifyVolumeName(volumeName);
+    verifyBucketName(bucketName);
+    Preconditions.checkNotNull(owner);
+    OmBucketArgs.Builder builder = OmBucketArgs.newBuilder();
+    builder.setVolumeName(volumeName)
+        .setBucketName(bucketName)
+        .setOwnerName(owner);
+    return ozoneManagerClient.setBucketOwner(builder.build());
+  }
 }

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

@@ -51,6 +51,10 @@ public final class OmBucketArgs extends WithMetadata implements Auditable {
 
   private long quotaInBytes;
   private long quotaInNamespace;
+  /**
+   * Bucket Owner Name.
+   */
+  private String ownerName;
 
   /**
    * Private constructor, constructed via builder.
@@ -61,16 +65,19 @@ public final class OmBucketArgs extends WithMetadata implements Auditable {
    * @param quotaInBytes Volume quota in bytes.
    * @param quotaInNamespace Volume quota in counts.
    */
+  @SuppressWarnings("checkstyle:ParameterNumber")
   private OmBucketArgs(String volumeName, String bucketName,
       Boolean isVersionEnabled, StorageType storageType,
-      Map<String, String> metadata, long quotaInBytes, long quotaInNamespace) {
+      Map<String, String> metadata, long quotaInBytes, long quotaInNamespace,
+      String ownerName) {
     this.volumeName = volumeName;
     this.bucketName = bucketName;
     this.isVersionEnabled = isVersionEnabled;
     this.storageType = storageType;
     this.metadata = metadata;
     this.quotaInBytes = quotaInBytes;
     this.quotaInNamespace = quotaInNamespace;
+    this.ownerName = ownerName;
   }
 
   /**
@@ -121,6 +128,14 @@ public long getQuotaInNamespace() {
     return quotaInNamespace;
   }
 
+  /**
+   * Returns Bucket Owner Name.
+   * @return ownerName.
+   */
+  public String getOwnerName() {
+    return ownerName;
+  }
+
   /**
    * Returns new builder class that builds a OmBucketArgs.
    * @return Builder
@@ -141,6 +156,9 @@ public Map<String, String> toAuditMap() {
     if(this.storageType != null){
       auditMap.put(OzoneConsts.STORAGE_TYPE, this.storageType.name());
     }
+    if (this.ownerName != null) {
+      auditMap.put(OzoneConsts.OWNER, this.ownerName);
+    }
     return auditMap;
   }
 
@@ -155,7 +173,7 @@ public static class Builder {
     private Map<String, String> metadata;
     private long quotaInBytes;
     private long quotaInNamespace;
-
+    private String ownerName;
     /**
      * Constructs a builder.
      */
@@ -199,6 +217,11 @@ public Builder setQuotaInNamespace(long quota) {
       return this;
     }
 
+    public Builder setOwnerName(String owner) {
+      ownerName = owner;
+      return this;
+    }
+
     /**
      * Constructs the OmBucketArgs.
      * @return instance of OmBucketArgs.
@@ -207,7 +230,7 @@ public OmBucketArgs build() {
       Preconditions.checkNotNull(volumeName);
       Preconditions.checkNotNull(bucketName);
       return new OmBucketArgs(volumeName, bucketName, isVersionEnabled,
-          storageType, metadata, quotaInBytes, quotaInNamespace);
+          storageType, metadata, quotaInBytes, quotaInNamespace, ownerName);
     }
   }
 
@@ -230,6 +253,9 @@ public BucketArgs getProtobuf() {
     if(quotaInNamespace > 0 || quotaInNamespace == OzoneConsts.QUOTA_RESET) {
       builder.setQuotaInNamespace(quotaInNamespace);
     }
+    if (ownerName != null) {
+      builder.setOwnerName(ownerName);
+    }
     return builder.build();
   }
 
@@ -247,6 +273,8 @@ public static OmBucketArgs getFromProtobuf(BucketArgs bucketArgs) {
             bucketArgs.getStorageType()) : null,
         KeyValueUtil.getFromProtobuf(bucketArgs.getMetadataList()),
         bucketArgs.getQuotaInBytes(),
-        bucketArgs.getQuotaInNamespace());
+        bucketArgs.getQuotaInNamespace(),
+        bucketArgs.hasOwnerName() ?
+            bucketArgs.getOwnerName() : null);
   }
 }

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

@@ -91,7 +91,7 @@ public final class OmBucketInfo extends WithObjectID implements Auditable {
    */
   private BucketLayout bucketLayout;
 
-  private final String owner;
+  private String owner;
 
   /**
    * Private constructor, constructed via builder.
@@ -297,6 +297,14 @@ public String getOwner() {
     return owner;
   }
 
+  public void setModificationTime(long modificationTime) {
+    this.modificationTime = modificationTime;
+  }
+
+  public void setOwner(String ownerName) {
+    this.owner = ownerName;
+  }
+
   /**
    * Returns new builder class that builds a OmBucketInfo.
    *

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java

@@ -199,6 +199,18 @@ default void setBucketProperty(OmBucketArgs args) throws IOException {
         "this to be implemented, as write requests use a new approach.");
   }
 
+  /**
+   * Changes the owner of a bucket.
+   * @param args  - OMBucketArgs
+   * @return true if operation succeeded, false if specified user is
+   *         already the owner.
+   * @throws IOException
+   */
+  default boolean setBucketOwner(OmBucketArgs args) throws IOException {
+    throw new UnsupportedOperationException("OzoneManager does not require " +
+        "this to be implemented, as write requests use a new approach.");
+  }
+
 
   /**
    * Open the given key and return an open key session.

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -526,6 +526,28 @@ public void setBucketProperty(OmBucketArgs args)
 
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public boolean setBucketOwner(OmBucketArgs args)
+      throws IOException {
+    SetBucketPropertyRequest.Builder req =
+        SetBucketPropertyRequest.newBuilder();
+    BucketArgs bucketArgs = args.getProtobuf();
+    req.setBucketArgs(bucketArgs);
+
+    OMRequest omRequest = createOMRequest(Type.SetBucketProperty)
+        .setSetBucketPropertyRequest(req)
+        .build();
+
+    OMResponse omResponse = submitRequest(omRequest);
+    SetBucketPropertyResponse response =
+        handleError(omResponse).getSetBucketPropertyResponse();
+
+    return response.getResponse();
+  }
+
   /**
    * List buckets in a volume.
    *

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

@@ -297,6 +297,28 @@ public void testVolumeSetOwner() throws IOException {
     proxy.setVolumeOwner(volumeName, ownerName);
   }
 
+  @Test
+  public void testBucketSetOwner() throws IOException {
+    String volumeName = UUID.randomUUID().toString();
+    String bucketName = UUID.randomUUID().toString();
+    store.createVolume(volumeName);
+    store.getVolume(volumeName).createBucket(bucketName);
+
+    String oldOwner = store.getVolume(volumeName).getBucket(bucketName)
+        .getOwner();
+    String ownerName = "testUser";
+
+    ClientProtocol proxy = store.getClientProxy();
+    proxy.setBucketOwner(volumeName, bucketName, ownerName);
+    String newOwner = store.getVolume(volumeName).getBucket(bucketName)
+        .getOwner();
+
+    assertEquals(ownerName, newOwner);
+    assertNotEquals(oldOwner, newOwner);
+    store.getVolume(volumeName).deleteBucket(bucketName);
+    store.deleteVolume(volumeName);
+  }
+
   @Test
   public void testSetAndClrQuota() throws Exception {
     String volumeName = UUID.randomUUID().toString();

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -623,6 +623,7 @@ message BucketArgs {
     repeated hadoop.hdds.KeyValue metadata = 7;
     optional uint64 quotaInBytes = 8;
     optional uint64 quotaInNamespace = 9;
+    optional string ownerName = 10;
 }
 
 message PrefixInfo {
@@ -730,7 +731,7 @@ message SetBucketPropertyRequest {
 }
 
 message SetBucketPropertyResponse {
-
+    optional bool response = 1;
 }
 
 message DeleteBucketRequest {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java

@@ -39,6 +39,7 @@
 import org.apache.hadoop.ozone.om.request.OMKeyRequestFactory;
 import org.apache.hadoop.ozone.om.request.bucket.OMBucketCreateRequest;
 import org.apache.hadoop.ozone.om.request.bucket.OMBucketDeleteRequest;
+import org.apache.hadoop.ozone.om.request.bucket.OMBucketSetOwnerRequest;
 import org.apache.hadoop.ozone.om.request.bucket.OMBucketSetPropertyRequest;
 import org.apache.hadoop.ozone.om.request.OMClientRequest;
 import org.apache.hadoop.ozone.om.request.bucket.acl.OMBucketAddAclRequest;
@@ -141,7 +142,13 @@ public static OMClientRequest createClientRequest(OMRequest omRequest,
     case DeleteBucket:
       return new OMBucketDeleteRequest(omRequest);
     case SetBucketProperty:
-      return new OMBucketSetPropertyRequest(omRequest);
+      boolean hasBucketOwner = omRequest.getSetBucketPropertyRequest()
+          .getBucketArgs().hasOwnerName();
+      if (hasBucketOwner) {
+        return new OMBucketSetOwnerRequest(omRequest);
+      } else {
+        return new OMBucketSetPropertyRequest(omRequest);
+      }
     case AddAcl:
     case RemoveAcl:
     case SetAcl: