HDDS-1393. Convert all OM Bucket related operations to HA model. (#704)

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

@@ -18,6 +18,8 @@
 
 package org.apache.hadoop.ozone.om.protocol;
 
+import org.apache.hadoop.ozone.om.helpers.OmBucketArgs;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
 import org.apache.hadoop.ozone.om.helpers.OmDeleteVolumeResponse;
 import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
@@ -166,4 +168,54 @@ public interface OzoneManagerHAProtocol {
    */
   void applyDeleteVolume(String volume, String owner,
       VolumeList newVolumeList) throws IOException;
+
+  /**
+   * Start Create Bucket Transaction.
+   * @param omBucketInfo
+   * @return OmBucketInfo
+   * @throws IOException
+   */
+  OmBucketInfo startCreateBucket(OmBucketInfo omBucketInfo) throws IOException;
+
+  /**
+   * Apply Create Bucket Changes to OM DB.
+   * @param omBucketInfo
+   * @throws IOException
+   */
+  void applyCreateBucket(OmBucketInfo omBucketInfo) throws IOException;
+
+  /**
+   * Start Delete Bucket Transaction.
+   * @param volumeName
+   * @param bucketName
+   * @throws IOException
+   */
+  void startDeleteBucket(String volumeName, String bucketName)
+      throws IOException;
+
+  /**
+   * Apply Delete Bucket changes to OM DB.
+   * @param volumeName
+   * @param bucketName
+   * @throws IOException
+   */
+  void applyDeleteBucket(String volumeName, String bucketName)
+      throws IOException;
+
+  /**
+   * Start SetBucket Property Transaction.
+   * @param omBucketArgs
+   * @return OmBucketInfo
+   * @throws IOException
+   */
+  OmBucketInfo startSetBucketProperty(OmBucketArgs omBucketArgs)
+      throws IOException;
+
+  /**
+   * Apply SetBucket Property changes to OM DB.
+   * @param omBucketInfo
+   * @throws IOException
+   */
+  void applySetBucketProperty(OmBucketInfo omBucketInfo) throws IOException;
+
 }

hadoop-ozone/common/src/main/proto/OzoneManagerProtocol.proto

@@ -478,7 +478,11 @@ message InfoBucketResponse {
 }
 
 message SetBucketPropertyRequest {
-    required BucketArgs bucketArgs = 1;
+    //TODO: See if we can merge bucketArgs and bucketInfo
+    optional BucketArgs bucketArgs = 1;
+    // This will be set during startTransaction, and used to apply to OM DB
+    // during applyTransaction.
+    optional BucketInfo bucketInfo = 2;
 }
 
 message SetBucketPropertyResponse {

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmMetrics.java

@@ -171,11 +171,11 @@ public class TestOmMetrics {
     Mockito.doNothing().when(mockS3Bm).deleteS3Bucket("random");
     Mockito.doReturn(true).when(mockS3Bm).createOzoneVolumeIfNeeded(null);
 
-    Mockito.doNothing().when(mockBm).createBucket(null);
-    Mockito.doNothing().when(mockBm).createBucket(null);
+    Mockito.doReturn(null).when(mockBm).createBucket(null);
+    Mockito.doReturn(null).when(mockBm).createBucket(null);
     Mockito.doNothing().when(mockBm).deleteBucket(null, null);
     Mockito.doReturn(null).when(mockBm).getBucketInfo(null, null);
-    Mockito.doNothing().when(mockBm).setBucketProperty(null);
+    Mockito.doReturn(null).when(mockBm).setBucketProperty(null);
     Mockito.doReturn(null).when(mockBm).listBuckets(null, null, null, 0);
 
     HddsWhiteboxTestUtils.setInternalState(

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java

@@ -20,11 +20,13 @@ import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.hadoop.hdds.client.ReplicationFactor;
 import org.apache.hadoop.hdds.client.ReplicationType;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdfs.LogVerificationAppender;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.MiniOzoneHAClusterImpl;
 import org.apache.hadoop.ozone.OzoneTestUtils;
+import org.apache.hadoop.ozone.client.BucketArgs;
 import org.apache.hadoop.ozone.client.ObjectStore;
 import org.apache.hadoop.ozone.client.OzoneBucket;
 import org.apache.hadoop.ozone.client.OzoneClient;
@@ -38,6 +40,7 @@ import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.ozone.client.OzoneClientFactory;
 import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.VolumeArgs;
+import org.apache.hadoop.util.Time;
 import org.apache.log4j.Logger;
 import org.junit.After;
 import org.junit.Assert;
@@ -125,23 +128,32 @@ public class TestOzoneManagerHA {
     }
   }
 
-  @Test
-  public void testAllVolumeOperations() throws Exception {
+
+  private OzoneVolume createAndCheckVolume(String volumeName)
+      throws Exception {
     String userName = "user" + RandomStringUtils.randomNumeric(5);
     String adminName = "admin" + RandomStringUtils.randomNumeric(5);
-    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
-
     VolumeArgs createVolumeArgs = VolumeArgs.newBuilder()
         .setOwner(userName)
         .setAdmin(adminName)
         .build();
 
     objectStore.createVolume(volumeName, createVolumeArgs);
-    OzoneVolume retVolumeinfo = objectStore.getVolume(volumeName);
 
-    Assert.assertTrue(retVolumeinfo.getName().equals(volumeName));
-    Assert.assertTrue(retVolumeinfo.getOwner().equals(userName));
-    Assert.assertTrue(retVolumeinfo.getAdmin().equals(adminName));
+    OzoneVolume retVolume = objectStore.getVolume(volumeName);
+
+    Assert.assertTrue(retVolume.getName().equals(volumeName));
+    Assert.assertTrue(retVolume.getOwner().equals(userName));
+    Assert.assertTrue(retVolume.getAdmin().equals(adminName));
+
+    return retVolume;
+  }
+  @Test
+  public void testAllVolumeOperations() throws Exception {
+
+    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
+
+    createAndCheckVolume(volumeName);
 
     objectStore.deleteVolume(volumeName);
 
@@ -152,6 +164,47 @@ public class TestOzoneManagerHA {
         () -> objectStore.deleteVolume(volumeName));
   }
 
+
+  @Test
+  public void testAllBucketOperations() throws Exception {
+
+    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
+    String bucketName = "volume" + RandomStringUtils.randomNumeric(5);
+
+    OzoneVolume retVolume = createAndCheckVolume(volumeName);
+
+    BucketArgs bucketArgs =
+        BucketArgs.newBuilder().setStorageType(StorageType.DISK)
+            .setVersioning(true).build();
+
+
+    retVolume.createBucket(bucketName, bucketArgs);
+
+
+    OzoneBucket ozoneBucket = retVolume.getBucket(bucketName);
+
+    Assert.assertEquals(volumeName, ozoneBucket.getVolumeName());
+    Assert.assertEquals(bucketName, ozoneBucket.getName());
+    Assert.assertTrue(ozoneBucket.getVersioning());
+    Assert.assertEquals(StorageType.DISK, ozoneBucket.getStorageType());
+    Assert.assertTrue(ozoneBucket.getCreationTime() <= Time.now());
+
+
+    // Change versioning to false
+    ozoneBucket.setVersioning(false);
+
+    ozoneBucket = retVolume.getBucket(bucketName);
+    Assert.assertFalse(ozoneBucket.getVersioning());
+
+    retVolume.deleteBucket(bucketName);
+
+    OzoneTestUtils.expectOmException(OMException.ResultCodes.BUCKET_NOT_FOUND,
+        () -> retVolume.deleteBucket(bucketName));
+
+
+
+  }
+
   /**
    * Test a client request when all OM nodes are running. The request should
    * succeed.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/BucketManager.java

@@ -30,7 +30,16 @@ public interface BucketManager {
    * Creates a bucket.
    * @param bucketInfo - OmBucketInfo for creating bucket.
    */
-  void createBucket(OmBucketInfo bucketInfo) throws IOException;
+  OmBucketInfo createBucket(OmBucketInfo bucketInfo) throws IOException;
+
+  /**
+   * Apply Create Bucket changes to OM DB.
+   * @param omBucketInfo
+   * @throws IOException
+   */
+  void applyCreateBucket(OmBucketInfo omBucketInfo) throws IOException;
+
+
   /**
    * Returns Bucket Information.
    * @param volumeName - Name of the Volume.
@@ -44,7 +53,14 @@ public interface BucketManager {
    * @param args - BucketArgs.
    * @throws IOException
    */
-  void setBucketProperty(OmBucketArgs args) throws IOException;
+  OmBucketInfo setBucketProperty(OmBucketArgs args) throws IOException;
+
+  /**
+   * Apply SetBucket Property changes to OM DB.
+   * @param omBucketInfo
+   * @throws IOException
+   */
+  void applySetBucketProperty(OmBucketInfo omBucketInfo) throws IOException;
 
   /**
    * Deletes an existing empty bucket from volume.
@@ -54,6 +70,15 @@ public interface BucketManager {
    */
   void deleteBucket(String volumeName, String bucketName) throws IOException;
 
+  /**
+   * Apply Delete Bucket changes to OM DB.
+   * @param volumeName
+   * @param bucketName
+   * @throws IOException
+   */
+  void applyDeleteBucket(String volumeName, String bucketName)
+      throws IOException;
+
   /**
    * Returns a list of buckets represented by {@link OmBucketInfo}
    * in the given volume.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/BucketManagerImpl.java

@@ -50,19 +50,27 @@ public class BucketManagerImpl implements BucketManager {
   private final OMMetadataManager metadataManager;
   private final KeyProviderCryptoExtension kmsProvider;
 
+  private final boolean isRatisEnabled;
+
   /**
    * Constructs BucketManager.
    *
    * @param metadataManager
    */
   public BucketManagerImpl(OMMetadataManager metadataManager) {
-    this(metadataManager, null);
+    this(metadataManager, null, false);
   }
 
   public BucketManagerImpl(OMMetadataManager metadataManager,
                            KeyProviderCryptoExtension kmsProvider) {
+    this(metadataManager, kmsProvider, false);
+  }
+
+  public BucketManagerImpl(OMMetadataManager metadataManager,
+      KeyProviderCryptoExtension kmsProvider, boolean isRatisEnabled) {
     this.metadataManager = metadataManager;
     this.kmsProvider = kmsProvider;
+    this.isRatisEnabled = isRatisEnabled;
   }
 
   KeyProviderCryptoExtension getKMSProvider() {
@@ -94,7 +102,7 @@ public class BucketManagerImpl implements BucketManager {
    * @param bucketInfo - OmBucketInfo.
    */
   @Override
-  public void createBucket(OmBucketInfo bucketInfo) throws IOException {
+  public OmBucketInfo createBucket(OmBucketInfo bucketInfo) throws IOException {
     Preconditions.checkNotNull(bucketInfo);
     String volumeName = bucketInfo.getVolumeName();
     String bucketName = bucketInfo.getBucketName();
@@ -155,9 +163,13 @@ public class BucketManagerImpl implements BucketManager {
       if (bekb != null) {
         omBucketInfoBuilder.setBucketEncryptionKey(bekb.build());
       }
-      metadataManager.getBucketTable().put(bucketKey,
-          omBucketInfoBuilder.build());
+
+      OmBucketInfo omBucketInfo = omBucketInfoBuilder.build();
+      if (!isRatisEnabled) {
+        commitCreateBucketInfoToDB(omBucketInfo);
+      }
       LOG.debug("created bucket: {} in volume: {}", bucketName, volumeName);
+      return omBucketInfo;
     } catch (IOException | DBException ex) {
       if (!(ex instanceof OMException)) {
         LOG.error("Bucket creation failed for bucket:{} in volume:{}",
@@ -170,6 +182,27 @@ public class BucketManagerImpl implements BucketManager {
     }
   }
 
+
+  public void applyCreateBucket(OmBucketInfo omBucketInfo) throws IOException {
+    Preconditions.checkNotNull(omBucketInfo);
+    try {
+      commitCreateBucketInfoToDB(omBucketInfo);
+    } catch (IOException ex) {
+      LOG.error("Apply CreateBucket Failed for bucket: {}, volume: {}",
+          omBucketInfo.getBucketName(), omBucketInfo.getVolumeName(), ex);
+      throw ex;
+    }
+  }
+
+  private void commitCreateBucketInfoToDB(OmBucketInfo omBucketInfo)
+      throws IOException {
+    String dbBucketKey =
+        metadataManager.getBucketKey(omBucketInfo.getVolumeName(),
+            omBucketInfo.getBucketName());
+    metadataManager.getBucketTable().put(dbBucketKey,
+        omBucketInfo);
+  }
+
   /**
    * Returns Bucket Information.
    *
@@ -210,7 +243,7 @@ public class BucketManagerImpl implements BucketManager {
    * @throws IOException - On Failure.
    */
   @Override
-  public void setBucketProperty(OmBucketArgs args) throws IOException {
+  public OmBucketInfo setBucketProperty(OmBucketArgs args) throws IOException {
     Preconditions.checkNotNull(args);
     String volumeName = args.getVolumeName();
     String bucketName = args.getBucketName();
@@ -262,8 +295,12 @@ public class BucketManagerImpl implements BucketManager {
       }
       bucketInfoBuilder.setCreationTime(oldBucketInfo.getCreationTime());
 
-      metadataManager.getBucketTable()
-          .put(bucketKey, bucketInfoBuilder.build());
+      OmBucketInfo omBucketInfo = bucketInfoBuilder.build();
+
+      if (!isRatisEnabled) {
+        commitSetBucketPropertyInfoToDB(omBucketInfo);
+      }
+      return omBucketInfo;
     } catch (IOException | DBException ex) {
       if (!(ex instanceof OMException)) {
         LOG.error("Setting bucket property failed for bucket:{} in volume:{}",
@@ -275,6 +312,23 @@ public class BucketManagerImpl implements BucketManager {
     }
   }
 
+  public void applySetBucketProperty(OmBucketInfo omBucketInfo)
+      throws IOException {
+    try {
+      commitSetBucketPropertyInfoToDB(omBucketInfo);
+    } catch (IOException ex) {
+      LOG.error("Apply SetBucket property failed for bucket:{} in " +
+              "volume:{}", omBucketInfo.getBucketName(),
+          omBucketInfo.getVolumeName(), ex);
+      throw ex;
+    }
+  }
+
+  private void commitSetBucketPropertyInfoToDB(OmBucketInfo omBucketInfo)
+      throws IOException {
+    commitCreateBucketInfoToDB(omBucketInfo);
+  }
+
   /**
    * Updates the existing ACL list with remove and add ACLs that are passed.
    * Remove is done before Add.
@@ -323,7 +377,10 @@ public class BucketManagerImpl implements BucketManager {
         throw new OMException("Bucket is not empty",
             OMException.ResultCodes.BUCKET_NOT_EMPTY);
       }
-      metadataManager.getBucketTable().delete(bucketKey);
+
+      if (!isRatisEnabled) {
+        commitDeleteBucketInfoToOMDB(bucketKey);
+      }
     } catch (IOException ex) {
       if (!(ex instanceof OMException)) {
         LOG.error("Delete bucket failed for bucket:{} in volume:{}", bucketName,
@@ -335,6 +392,25 @@ public class BucketManagerImpl implements BucketManager {
     }
   }
 
+  public void applyDeleteBucket(String volumeName, String bucketName)
+      throws IOException {
+    Preconditions.checkNotNull(volumeName);
+    Preconditions.checkNotNull(bucketName);
+    try {
+      commitDeleteBucketInfoToOMDB(metadataManager.getBucketKey(volumeName,
+          bucketName));
+    } catch (IOException ex) {
+      LOG.error("Apply DeleteBucket Failed for bucket: {}, volume: {}",
+          bucketName, volumeName, ex);
+      throw ex;
+    }
+  }
+
+  private void commitDeleteBucketInfoToOMDB(String dbBucketKey)
+      throws IOException {
+    metadataManager.getBucketTable().delete(dbBucketKey);
+  }
+
   /**
    * {@inheritDoc}
    */

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -279,6 +279,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
   private static String keyProviderUriKeyName =
       CommonConfigurationKeysPublic.HADOOP_SECURITY_KEY_PROVIDER_PATH;
 
+
   private OzoneManager(OzoneConfiguration conf) throws IOException,
       AuthenticationException {
     super(OzoneVersionInfo.OZONE_VERSION_INFO);
@@ -335,7 +336,8 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
       LOG.error("Fail to create Key Provider");
     }
 
-    bucketManager = new BucketManagerImpl(metadataManager, getKmsProvider());
+    bucketManager = new BucketManagerImpl(metadataManager, getKmsProvider(),
+        isRatisEnabled);
     metrics = OMMetrics.create();
 
     s3BucketManager = new S3BucketManagerImpl(configuration, metadataManager,
@@ -1645,6 +1647,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
     return volumeList;
   }
 
+  @Override
   public void applyCreateVolume(OmVolumeArgs omVolumeArgs,
       VolumeList volumeList) throws IOException {
     // TODO: Need to add metrics and Audit log for HA requests
@@ -1707,6 +1710,66 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
   }
 
 
+  @Override
+  public OmBucketInfo startCreateBucket(OmBucketInfo omBucketInfo)
+      throws IOException {
+    Preconditions.checkNotNull(omBucketInfo);
+    if(isAclEnabled) {
+      checkAcls(ResourceType.BUCKET, StoreType.OZONE, ACLType.CREATE,
+          omBucketInfo.getVolumeName(), omBucketInfo.getBucketName(), null);
+    }
+
+    return bucketManager.createBucket(omBucketInfo);
+  }
+
+  @Override
+  public void applyCreateBucket(OmBucketInfo omBucketInfo) throws IOException {
+    // TODO: Need to add metrics and Audit log for HA requests
+    bucketManager.applyCreateBucket(omBucketInfo);
+  }
+
+
+  @Override
+  public void startDeleteBucket(String volumeName, String bucketName)
+      throws IOException {
+    // TODO: Need to add metrics and Audit log for HA requests
+    if(isAclEnabled) {
+      checkAcls(ResourceType.BUCKET, StoreType.OZONE, ACLType.CREATE,
+          volumeName, bucketName, null);
+    }
+
+    bucketManager.deleteBucket(volumeName, bucketName);
+  }
+
+
+  @Override
+  public void applyDeleteBucket(String volumeName, String bucketName)
+      throws IOException {
+    // TODO: Need to add metrics and Audit log for HA requests
+    bucketManager.applyDeleteBucket(volumeName, bucketName);
+  }
+
+
+  @Override
+  public OmBucketInfo startSetBucketProperty(OmBucketArgs omBucketArgs)
+      throws IOException {
+    Preconditions.checkNotNull(omBucketArgs);
+    // TODO: Need to add metrics and Audit log for HA requests
+    if(isAclEnabled) {
+      checkAcls(ResourceType.BUCKET, StoreType.OZONE, ACLType.CREATE,
+          omBucketArgs.getVolumeName(), omBucketArgs.getBucketName(), null);
+    }
+    return bucketManager.setBucketProperty(omBucketArgs);
+  }
+
+
+  @Override
+  public void applySetBucketProperty(OmBucketInfo omBucketInfo)
+      throws IOException {
+    // TODO: Need to add metrics and Audit log for HA requests
+    bucketManager.applySetBucketProperty(omBucketInfo);
+  }
+
   /**
    * Checks if current caller has acl permissions.
    *

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/S3BucketManagerImpl.java

@@ -141,7 +141,14 @@ public class S3BucketManagerImpl implements S3BucketManager {
         throw new OMException("No such S3 bucket. " + bucketName,
             OMException.ResultCodes.S3_BUCKET_NOT_FOUND);
       }
-      bucketManager.deleteBucket(getOzoneVolumeName(bucketName), bucketName);
+
+      if (isRatisEnabled) {
+        bucketManager.deleteBucket(getOzoneVolumeName(bucketName), bucketName);
+        bucketManager.applyDeleteBucket(getOzoneVolumeName(bucketName),
+            bucketName);
+      } else {
+        bucketManager.deleteBucket(getOzoneVolumeName(bucketName), bucketName);
+      }
       omMetadataManager.getS3Table().delete(bucket);
     } catch(IOException ex) {
       throw ex;
@@ -200,7 +207,11 @@ public class S3BucketManagerImpl implements S3BucketManager {
             .setIsVersionEnabled(Boolean.FALSE)
             .setStorageType(StorageType.DEFAULT)
             .build();
-    bucketManager.createBucket(bucketInfo);
+    if (isRatisEnabled) {
+      bucketManager.applyCreateBucket(bucketManager.createBucket(bucketInfo));
+    } else {
+      bucketManager.createBucket(bucketInfo);
+    }
   }
 
   @Override

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

@@ -192,6 +192,9 @@ public class OzoneManagerStateMachine extends BaseStateMachine {
       case CreateVolume:
       case SetVolumeProperty:
       case DeleteVolume:
+      case CreateBucket:
+      case SetBucketProperty:
+      case DeleteBucket:
         newOmRequest = handler.handleStartTransaction(omRequest);
         break;
       case AllocateBlock:

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerHARequestHandlerImpl.java

@@ -19,15 +19,25 @@ package org.apache.hadoop.ozone.protocolPB;
 
 import java.io.IOException;
 
+import org.apache.hadoop.ozone.om.helpers.OmBucketArgs;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
 import org.apache.hadoop.ozone.om.helpers.OmDeleteVolumeResponse;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeOwnerChangeResponse;
 import org.apache.hadoop.ozone.om.protocol.OzoneManagerServerProtocol;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .CreateBucketRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .CreateBucketResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
     .CreateVolumeRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
     .CreateVolumeResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .DeleteBucketRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .DeleteBucketResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
     .DeleteVolumeRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
@@ -36,6 +46,10 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
     .OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
     .OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .SetBucketPropertyRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
+    .SetBucketPropertyResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
     .SetVolumePropertyRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
@@ -76,6 +90,15 @@ public class OzoneManagerHARequestHandlerImpl
     case DeleteVolume:
       newOmRequest = handleDeleteVolumeStart(omRequest);
       break;
+    case CreateBucket:
+      newOmRequest = handleCreateBucketStart(omRequest);
+      break;
+    case SetBucketProperty:
+      newOmRequest = handleSetBucketPropertyStart(omRequest);
+      break;
+    case DeleteBucket:
+      newOmRequest = handleDeleteBucketRequestStart(omRequest);
+      break;
     default:
       throw new IOException("Unrecognized Command Type:" + cmdType);
     }
@@ -104,6 +127,18 @@ public class OzoneManagerHARequestHandlerImpl
         responseBuilder.setDeleteVolumeResponse(
             handleDeleteVolumeApply(omRequest));
         break;
+      case CreateBucket:
+        responseBuilder.setCreateBucketResponse(
+            handleCreateBucketApply(omRequest));
+        break;
+      case SetBucketProperty:
+        responseBuilder.setSetBucketPropertyResponse(
+            handleSetBucketPropertyApply(omRequest));
+        break;
+      case DeleteBucket:
+        responseBuilder.setDeleteBucketResponse(
+            handleDeleteBucketApply(omRequest));
+        break;
       default:
         // As all request types are not changed so we need to call handle
         // here.
@@ -244,4 +279,90 @@ public class OzoneManagerHARequestHandlerImpl
     return DeleteVolumeResponse.newBuilder().build();
   }
 
+  private OMRequest handleCreateBucketStart(OMRequest omRequest)
+      throws IOException {
+
+    CreateBucketRequest createBucketRequest =
+        omRequest.getCreateBucketRequest();
+
+    OmBucketInfo omBucketInfo =
+        getOzoneManagerServerProtocol().startCreateBucket(
+        OmBucketInfo.getFromProtobuf(createBucketRequest.getBucketInfo()));
+
+    CreateBucketRequest newCreateBucketRequest =
+        CreateBucketRequest.newBuilder().setBucketInfo(
+            omBucketInfo.getProtobuf()).build();
+    return omRequest.toBuilder().setCreateBucketRequest(newCreateBucketRequest)
+        .build();
+
+  }
+
+
+  private CreateBucketResponse handleCreateBucketApply(OMRequest omRequest)
+      throws IOException {
+    CreateBucketRequest createBucketRequest =
+        omRequest.getCreateBucketRequest();
+
+    getOzoneManagerServerProtocol().applyCreateBucket(
+        OmBucketInfo.getFromProtobuf(createBucketRequest.getBucketInfo()));
+
+    return CreateBucketResponse.newBuilder().build();
+  }
+
+
+  private OMRequest handleDeleteBucketRequestStart(OMRequest omRequest)
+      throws IOException {
+
+    DeleteBucketRequest deleteBucketRequest =
+        omRequest.getDeleteBucketRequest();
+    getOzoneManagerServerProtocol().startDeleteBucket(
+        deleteBucketRequest.getVolumeName(),
+        deleteBucketRequest.getBucketName());
+
+    return omRequest;
+  }
+
+  private DeleteBucketResponse handleDeleteBucketApply(OMRequest omRequest)
+      throws IOException {
+
+    DeleteBucketRequest deleteBucketRequest =
+        omRequest.getDeleteBucketRequest();
+
+    getOzoneManagerServerProtocol().applyDeleteBucket(
+        deleteBucketRequest.getVolumeName(),
+        deleteBucketRequest.getBucketName());
+
+    return DeleteBucketResponse.newBuilder().build();
+  }
+
+  private OMRequest handleSetBucketPropertyStart(
+      OMRequest omRequest) throws IOException {
+    SetBucketPropertyRequest setBucketPropertyRequest =
+        omRequest.getSetBucketPropertyRequest();
+
+    OmBucketInfo omBucketInfo =
+        getOzoneManagerServerProtocol().startSetBucketProperty(
+        OmBucketArgs.getFromProtobuf(setBucketPropertyRequest.getBucketArgs()));
+
+    SetBucketPropertyRequest newSetBucketPropertyRequest =
+        SetBucketPropertyRequest.newBuilder()
+            .setBucketInfo(omBucketInfo.getProtobuf()).build();
+
+    return omRequest.toBuilder().setSetBucketPropertyRequest(
+        newSetBucketPropertyRequest).build();
+  }
+
+  private SetBucketPropertyResponse handleSetBucketPropertyApply(
+      OMRequest omRequest) throws IOException {
+    SetBucketPropertyRequest setBucketPropertyRequest =
+        omRequest.getSetBucketPropertyRequest();
+
+    getOzoneManagerServerProtocol().applySetBucketProperty(
+        OmBucketInfo.getFromProtobuf(setBucketPropertyRequest.getBucketInfo()));
+
+    return SetBucketPropertyResponse.newBuilder().build();
+  }
+
+
+
 }