diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java
index 441d9143b598..ca885b3b6b06 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java
@@ -413,6 +413,12 @@ public void setListCacheSize(int listCacheSize) {
     this.listCacheSize = listCacheSize;
   }
 
+  @Deprecated
+  public void setEncryptionKey(String bekName) throws IOException {
+    proxy.setEncryptionKey(volumeName, name, bekName);
+    encryptionKeyName = bekName;
+  }
+
   /**
    * Creates a new key in the bucket, with default replication type RATIS and
    * with replication factor THREE.
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
index 5316f7a99e9f..e455e3040adb 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
@@ -997,6 +997,24 @@ void setBucketQuota(String volumeName, String bucketName,
   void setReplicationConfig(String volumeName, String bucketName,
       ReplicationConfig replicationConfig) throws IOException;
 
+  /**
+   * Set Bucket Encryption Key (BEK).
+   *
+   * @param volumeName
+   * @param bucketName
+   * @param bekName
+   * @throws IOException
+   * @deprecated This functionality is deprecated as it is not intended for
+   * users to reset bucket encryption under normal circumstances and may be
+   * removed in the future. Users are advised to exercise caution and consider
+   * alternative approaches for managing bucket encryption unless HDDS-7449 or
+   * HDDS-7526 is encountered. As a result, the setter methods for this
+   * functionality have been marked as deprecated.
+   */
+  @Deprecated
+  void setEncryptionKey(String volumeName, String bucketName,
+                        String bekName) throws IOException;
+
   /**
    * Returns OzoneKey that contains the application generated/visible
    * metadata for an Ozone Object.
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
index 850ae0d19376..e14ae5828d70 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
@@ -1213,6 +1213,22 @@ public void setBucketQuota(String volumeName, String bucketName,
 
   }
 
+  @Deprecated
+  @Override
+  public void setEncryptionKey(String volumeName, String bucketName,
+                               String bekName) throws IOException {
+    verifyVolumeName(volumeName);
+    verifyBucketName(bucketName);
+    OmBucketArgs.Builder builder = OmBucketArgs.newBuilder();
+    BucketEncryptionKeyInfo bek = new BucketEncryptionKeyInfo.Builder()
+        .setKeyName(bekName).build();
+    builder.setVolumeName(volumeName)
+        .setBucketName(bucketName)
+        .setBucketEncryptionKey(bek);
+    OmBucketArgs finalArgs = builder.build();
+    ozoneManagerClient.setBucketProperty(finalArgs);
+  }
+
   @Override
   public void setReplicationConfig(
       String volumeName, String bucketName, ReplicationConfig replicationConfig)
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmBucketArgs.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmBucketArgs.java
index f8c752aab271..e382377dff45 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmBucketArgs.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmBucketArgs.java
@@ -27,6 +27,7 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.BucketArgs;
 
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.ozone.protocolPB.OMPBHelper;
 
 /**
  * A class that encapsulates Bucket Arguments.
@@ -50,6 +51,10 @@ public final class OmBucketArgs extends WithMetadata implements Auditable {
    */
   private StorageType storageType;
 
+  /**
+   * Bucket encryption key info if encryption is enabled.
+   */
+  private BucketEncryptionKeyInfo bekInfo;
   private long quotaInBytes = OzoneConsts.QUOTA_RESET;
   private long quotaInNamespace = OzoneConsts.QUOTA_RESET;
   private boolean quotaInBytesSet = false;
@@ -150,6 +155,10 @@ public DefaultReplicationConfig getDefaultReplicationConfig() {
     return defaultReplicationConfig;
   }
 
+  public BucketEncryptionKeyInfo getBucketEncryptionKeyInfo() {
+    return bekInfo;
+  }
+
   /**
    * Sets the Bucket default replication config.
    */
@@ -168,6 +177,12 @@ private void setQuotaInNamespace(long quotaInNamespace) {
     this.quotaInNamespace = quotaInNamespace;
   }
 
+  @Deprecated
+  private void setBucketEncryptionKey(
+      BucketEncryptionKeyInfo bucketEncryptionKey) {
+    this.bekInfo = bucketEncryptionKey;
+  }
+
   /**
    * Returns Bucket Owner Name.
    *
@@ -216,6 +231,7 @@ public static class Builder {
     private long quotaInBytes;
     private boolean quotaInNamespaceSet = false;
     private long quotaInNamespace;
+    private BucketEncryptionKeyInfo bekInfo;
     private DefaultReplicationConfig defaultReplicationConfig;
     private String ownerName;
     /**
@@ -241,6 +257,12 @@ public Builder setIsVersionEnabled(Boolean versionFlag) {
       return this;
     }
 
+    @Deprecated
+    public Builder setBucketEncryptionKey(BucketEncryptionKeyInfo info) {
+      this.bekInfo = info;
+      return this;
+    }
+
     public Builder addMetadata(Map<String, String> metadataMap) {
       this.metadata = metadataMap;
       return this;
@@ -291,6 +313,9 @@ public OmBucketArgs build() {
       if (quotaInNamespaceSet) {
         omBucketArgs.setQuotaInNamespace(quotaInNamespace);
       }
+      if (bekInfo != null && bekInfo.getKeyName() != null) {
+        omBucketArgs.setBucketEncryptionKey(bekInfo);
+      }
       return omBucketArgs;
     }
   }
@@ -322,6 +347,11 @@ public BucketArgs getProtobuf() {
     if (ownerName != null) {
       builder.setOwnerName(ownerName);
     }
+
+    if (bekInfo != null && bekInfo.getKeyName() != null) {
+      builder.setBekInfo(OMPBHelper.convert(bekInfo));
+    }
+
     return builder.build();
   }
 
@@ -355,6 +385,11 @@ public static OmBucketArgs getFromProtobuf(BucketArgs bucketArgs) {
     if (bucketArgs.hasQuotaInNamespace()) {
       omBucketArgs.setQuotaInNamespace(bucketArgs.getQuotaInNamespace());
     }
+
+    if (bucketArgs.hasBekInfo()) {
+      omBucketArgs.setBucketEncryptionKey(
+          OMPBHelper.convert(bucketArgs.getBekInfo()));
+    }
     return omBucketArgs;
   }
 }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java
index 92381829f0ba..539243cbad6e 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java
@@ -29,12 +29,17 @@
 import java.util.List;
 import java.util.UUID;
 
+import org.apache.hadoop.crypto.key.KeyProvider;
+import org.apache.hadoop.crypto.key.kms.KMSClientProvider;
+import org.apache.hadoop.crypto.key.kms.server.MiniKMS;
 import org.apache.hadoop.hdds.utils.IOUtils;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.FileChecksum;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.TrashPolicy;
 import org.apache.hadoop.hdds.cli.GenericCli;
 import org.apache.hadoop.hdds.cli.OzoneAdmin;
 import org.apache.hadoop.hdds.client.ReplicationType;
@@ -60,7 +65,6 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.ozone.test.GenericTestUtils;
 import org.apache.hadoop.util.ToolRunner;
-import org.apache.hadoop.fs.TrashPolicy;
 import org.apache.hadoop.ozone.om.TrashPolicyOzone;
 
 import com.google.common.base.Strings;
@@ -81,6 +85,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -117,6 +122,8 @@ public class TestOzoneShellHA {
   private static File testFile;
   private static String testFilePathString;
   private static MiniOzoneCluster cluster = null;
+  private static File testDir;
+  private static MiniKMS miniKMS;
   private static OzoneClient client;
   private OzoneShell ozoneShell = null;
   private OzoneAdmin ozoneAdminShell = null;
@@ -140,9 +147,20 @@ public class TestOzoneShellHA {
   @BeforeAll
   public static void init() throws Exception {
     OzoneConfiguration conf = new OzoneConfiguration();
+    startKMS();
     startCluster(conf);
   }
 
+  protected static void startKMS() throws Exception {
+    testDir = GenericTestUtils.getTestDir(
+        TestOzoneShellHA.class.getSimpleName());
+    File kmsDir = new File(testDir, UUID.randomUUID().toString());
+    assertTrue(kmsDir.mkdirs());
+    MiniKMS.Builder miniKMSBuilder = new MiniKMS.Builder();
+    miniKMS = miniKMSBuilder.setKmsConfDir(kmsDir).build();
+    miniKMS.start();
+  }
+
   protected static void startCluster(OzoneConfiguration conf) throws Exception {
     String path = GenericTestUtils.getTempPath(
         TestOzoneShellHA.class.getSimpleName());
@@ -160,6 +178,8 @@ protected static void startCluster(OzoneConfiguration conf) throws Exception {
     clusterId = UUID.randomUUID().toString();
     scmId = UUID.randomUUID().toString();
     final int numDNs = 5;
+    conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_KEY_PROVIDER_PATH,
+        getKeyProviderURI(miniKMS));
     cluster = MiniOzoneCluster.newOMHABuilder(conf)
         .setClusterId(clusterId)
         .setScmId(scmId)
@@ -181,9 +201,17 @@ public static void shutdown() {
       cluster.shutdown();
     }
 
+    if (miniKMS != null) {
+      miniKMS.stop();
+    }
+
     if (baseDir != null) {
       FileUtil.fullyDelete(baseDir, true);
     }
+
+    if (testDir != null) {
+      FileUtil.fullyDelete(testDir, true);
+    }
   }
 
   @BeforeEach
@@ -1299,6 +1327,33 @@ public void testSetECReplicationConfigOnBucket() throws Exception {
     }
   }
 
+  @Test
+  public void testSetEncryptionKey() throws Exception {
+    final String volumeName = "volume111";
+    getVolume(volumeName);
+    String bucketPath = "/volume111/bucket0";
+    String[] args = new String[]{"bucket", "create", bucketPath};
+    execute(ozoneShell, args);
+
+    OzoneVolume volume =
+        client.getObjectStore().getVolume(volumeName);
+    OzoneBucket bucket = volume.getBucket("bucket0");
+    assertNull(bucket.getEncryptionKeyName());
+    String newEncKey = "enckey1";
+
+    KeyProvider provider = cluster.getOzoneManager().getKmsProvider();
+    KeyProvider.Options options = KeyProvider.options(cluster.getConf());
+    options.setDescription(newEncKey);
+    options.setBitLength(128);
+    provider.createKey(newEncKey, options);
+    provider.flush();
+
+    args = new String[]{"bucket", "set-encryption-key", bucketPath, "-k",
+        newEncKey};
+    execute(ozoneShell, args);
+    assertEquals(newEncKey, volume.getBucket("bucket0").getEncryptionKeyName());
+  }
+
   @Test
   public void testCreateBucketWithECReplicationConfigWithoutReplicationParam() {
     getVolume("volume102");
@@ -1912,4 +1967,9 @@ public void testLinkedAndNonLinkedBucketMetaData()
         new String[]{"volume", "delete", "/volume1"});
     out.reset();
   }
+
+  private static String getKeyProviderURI(MiniKMS kms) {
+    return KMSClientProvider.SCHEME_NAME + "://" +
+        kms.getKMSUrl().toExternalForm().replace("://", "@");
+  }
 }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHAWithFSO.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHAWithFSO.java
index 0c6a5e814380..3d1757ecbd9c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHAWithFSO.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHAWithFSO.java
@@ -37,6 +37,7 @@ public static void init() throws Exception {
     OzoneConfiguration conf = new OzoneConfiguration();
     conf.set(OMConfigKeys.OZONE_DEFAULT_BUCKET_LAYOUT,
         OMConfigKeys.OZONE_BUCKET_LAYOUT_FILE_SYSTEM_OPTIMIZED);
+    startKMS();
     startCluster(conf);
   }
 
diff --git a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
index e49f23b11528..71a59893f6f7 100644
--- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
+++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
@@ -805,6 +805,7 @@ message BucketArgs {
     optional uint64 quotaInNamespace = 9;
     optional string ownerName = 10;
     optional hadoop.hdds.DefaultReplicationConfig defaultReplicationConfig = 11;
+    optional BucketEncryptionInfoProto bekInfo = 12;
 }
 
 message PrefixInfo {
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/common/BekInfoUtils.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/common/BekInfoUtils.java
new file mode 100644
index 000000000000..7cfad3b8a33a
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/common/BekInfoUtils.java
@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership.  The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.ozone.common;
+
+import org.apache.hadoop.crypto.CipherSuite;
+import org.apache.hadoop.crypto.key.KeyProvider;
+import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.BucketEncryptionInfoProto;
+import org.apache.hadoop.ozone.protocolPB.OMPBHelper;
+
+import java.io.IOException;
+
+import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CryptoProtocolVersionProto.ENCRYPTION_ZONES;
+
+/**
+ * Utility class for common bucket encryption key operations.
+ */
+public final class BekInfoUtils {
+
+  private BekInfoUtils() {
+  }
+
+  public static BucketEncryptionInfoProto getBekInfo(
+      KeyProviderCryptoExtension kmsProvider, BucketEncryptionInfoProto bek)
+      throws IOException {
+    BucketEncryptionInfoProto.Builder bekb = null;
+    if (kmsProvider == null) {
+      throw new OMException("Invalid KMS provider, check configuration " +
+          CommonConfigurationKeys.HADOOP_SECURITY_KEY_PROVIDER_PATH,
+          OMException.ResultCodes.INVALID_KMS_PROVIDER);
+    }
+    if (bek.getKeyName() == null) {
+      throw new OMException("Bucket encryption key needed.", OMException
+          .ResultCodes.BUCKET_ENCRYPTION_KEY_NOT_FOUND);
+    }
+    // Talk to KMS to retrieve the bucket encryption key info.
+    KeyProvider.Metadata metadata = kmsProvider.getMetadata(
+        bek.getKeyName());
+    if (metadata == null) {
+      throw new OMException("Bucket encryption key " + bek.getKeyName()
+          + " doesn't exist.",
+          OMException.ResultCodes.BUCKET_ENCRYPTION_KEY_NOT_FOUND);
+    }
+    // If the provider supports pool for EDEKs, this will fill in the pool
+    kmsProvider.warmUpEncryptedKeys(bek.getKeyName());
+    bekb = BucketEncryptionInfoProto.newBuilder()
+        .setKeyName(bek.getKeyName())
+        .setCryptoProtocolVersion(ENCRYPTION_ZONES)
+        .setSuite(OMPBHelper.convert(
+            CipherSuite.convert(metadata.getCipher())));
+    return bekb.build();
+  }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketCreateRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketCreateRequest.java
index de8152af4687..7cce3ac456f9 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketCreateRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketCreateRequest.java
@@ -18,10 +18,7 @@
 
 package org.apache.hadoop.ozone.om.request.bucket;
 
-import org.apache.hadoop.crypto.CipherSuite;
-import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.hdds.client.DefaultReplicationConfig;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
@@ -33,6 +30,7 @@
 import org.apache.hadoop.ozone.audit.AuditLogger;
 import org.apache.hadoop.ozone.audit.OMAction;
 import org.apache.hadoop.ozone.om.OMConfigKeys;
+import org.apache.hadoop.ozone.common.BekInfoUtils;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OMMetrics;
 import org.apache.hadoop.ozone.om.OzoneManager;
@@ -52,14 +50,12 @@
 import org.apache.hadoop.ozone.om.response.bucket.OMBucketCreateResponse;
 import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.BucketEncryptionInfoProto;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.BucketInfo;
 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.OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
-import org.apache.hadoop.ozone.protocolPB.OMPBHelper;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.apache.hadoop.util.Time;
@@ -75,7 +71,6 @@
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.VOLUME_NOT_FOUND;
 import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.BUCKET_LOCK;
 import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.VOLUME_LOCK;
-import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CryptoProtocolVersionProto.ENCRYPTION_ZONES;
 
 /**
  * Handles CreateBucket Request.
@@ -116,7 +111,8 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
         .setModificationTime(initialTime);
 
     if (bucketInfo.hasBeinfo()) {
-      newBucketInfo.setBeinfo(getBeinfo(kmsProvider, bucketInfo));
+      newBucketInfo.setBeinfo(
+          BekInfoUtils.getBekInfo(kmsProvider, bucketInfo.getBeinfo()));
     }
 
     boolean hasSourceVolume = bucketInfo.hasSourceVolume();
@@ -338,38 +334,6 @@ private void addDefaultAcls(OmBucketInfo omBucketInfo,
     omBucketInfo.setAcls(acls);
   }
 
-  private BucketEncryptionInfoProto getBeinfo(
-      KeyProviderCryptoExtension kmsProvider, BucketInfo bucketInfo)
-      throws IOException {
-    BucketEncryptionInfoProto bek = bucketInfo.getBeinfo();
-    BucketEncryptionInfoProto.Builder bekb = null;
-    if (kmsProvider == null) {
-      throw new OMException("Invalid KMS provider, check configuration " +
-          CommonConfigurationKeys.HADOOP_SECURITY_KEY_PROVIDER_PATH,
-          OMException.ResultCodes.INVALID_KMS_PROVIDER);
-    }
-    if (bek.getKeyName() == null) {
-      throw new OMException("Bucket encryption key needed.", OMException
-          .ResultCodes.BUCKET_ENCRYPTION_KEY_NOT_FOUND);
-    }
-    // Talk to KMS to retrieve the bucket encryption key info.
-    KeyProvider.Metadata metadata = kmsProvider.getMetadata(
-        bek.getKeyName());
-    if (metadata == null) {
-      throw new OMException("Bucket encryption key " + bek.getKeyName()
-          + " doesn't exist.",
-          OMException.ResultCodes.BUCKET_ENCRYPTION_KEY_NOT_FOUND);
-    }
-    // If the provider supports pool for EDEKs, this will fill in the pool
-    kmsProvider.warmUpEncryptedKeys(bek.getKeyName());
-    bekb = BucketEncryptionInfoProto.newBuilder()
-        .setKeyName(bek.getKeyName())
-        .setCryptoProtocolVersion(ENCRYPTION_ZONES)
-        .setSuite(OMPBHelper.convert(
-            CipherSuite.convert(metadata.getCipher())));
-    return bekb.build();
-  }
-
   /**
    * Check namespace quota.
    */
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketSetPropertyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketSetPropertyRequest.java
index 821c374c2d10..9c7ef1087c10 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketSetPropertyRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketSetPropertyRequest.java
@@ -23,12 +23,15 @@
 import java.util.List;
 
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
 import org.apache.hadoop.hdds.client.DefaultReplicationConfig;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.audit.AuditLogger;
 import org.apache.hadoop.ozone.audit.OMAction;
+import org.apache.hadoop.ozone.common.BekInfoUtils;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.helpers.BucketEncryptionKeyInfo;
 import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
 import org.apache.hadoop.ozone.om.request.validation.RequestFeatureValidator;
 import org.apache.hadoop.ozone.om.request.validation.RequestProcessingPhase;
@@ -87,6 +90,18 @@ public OMRequest preExecute(OzoneManager ozoneManager)
         .getSetBucketPropertyRequest().toBuilder()
         .setModificationTime(modificationTime);
 
+    BucketArgs bucketArgs =
+        getOmRequest().getSetBucketPropertyRequest().getBucketArgs();
+
+    if (bucketArgs.hasBekInfo()) {
+      KeyProviderCryptoExtension kmsProvider = ozoneManager.getKmsProvider();
+      BucketArgs.Builder bucketArgsBuilder =
+          setBucketPropertyRequestBuilder.getBucketArgsBuilder();
+      bucketArgsBuilder.setBekInfo(
+          BekInfoUtils.getBekInfo(kmsProvider, bucketArgs.getBekInfo()));
+      setBucketPropertyRequestBuilder.setBucketArgs(bucketArgsBuilder.build());
+    }
+
     return getOmRequest().toBuilder()
         .setSetBucketPropertyRequest(setBucketPropertyRequestBuilder)
         .setUserInfo(getUserInfo())
@@ -190,6 +205,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
         bucketInfoBuilder.setDefaultReplicationConfig(defaultReplicationConfig);
       }
 
+      BucketEncryptionKeyInfo bek = omBucketArgs.getBucketEncryptionKeyInfo();
+      if (bek != null && bek.getKeyName() != null) {
+        bucketInfoBuilder.setBucketEncryptionKey(bek);
+      }
+
       omBucketInfo = bucketInfoBuilder.build();
 
       // Update table cache.
diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java
index 174af69e255d..f0528facbb67 100644
--- a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java
+++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java
@@ -573,6 +573,14 @@ public void setReplicationConfig(String volumeName, String bucketName,
 
   }
 
+  @Deprecated
+  @Override
+  public void setEncryptionKey(String volumeName, String bucketName,
+                               String bekName)
+      throws IOException {
+
+  }
+
   @Override
   public OzoneKey headObject(String volumeName, String bucketName,
                              String keyName) throws IOException {
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/bucket/BucketCommands.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/bucket/BucketCommands.java
index 454660e2ca32..80e26e044516 100644
--- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/bucket/BucketCommands.java
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/bucket/BucketCommands.java
@@ -50,7 +50,8 @@
         SetAclBucketHandler.class,
         ClearQuotaHandler.class,
         SetReplicationConfigHandler.class,
-        UpdateBucketHandler.class
+        UpdateBucketHandler.class,
+        SetEncryptionKey.class
     },
     mixinStandardHelpOptions = true,
     versionProvider = HddsVersionProvider.class)
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/bucket/SetEncryptionKey.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/bucket/SetEncryptionKey.java
new file mode 100644
index 000000000000..86a50e9df3c7
--- /dev/null
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/bucket/SetEncryptionKey.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package org.apache.hadoop.ozone.shell.bucket;
+
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.OzoneClientException;
+import org.apache.hadoop.ozone.shell.OzoneAddress;
+import picocli.CommandLine;
+
+import java.io.IOException;
+
+/**
+ * Command-line tool to set the encryption key of a bucket.
+ *
+ * There are known bugs, HDDS-7449 and HDDS-7526, which could potentially result
+ * in the loss of bucket encryption properties when either quota or bucket
+ * replication properties are (re)set on an existing bucket, posing a critical
+ * issue. This may affect consumers using previous versions of Ozone.
+ *
+ * To address this bug, this CLI tool provides the ability to (re)set the
+ * Bucket Encryption Key (BEK) for HDDS-7449/HDDS-7526 affected buckets using
+ * the Ozone shell.
+ *
+ * Users can execute the following command for setting BEK for a given bucket:
+ * "ozone sh bucket set-encryption-key -k <enckey> <vol>/<buck>"
+ *
+ * Please note that this operation only resets the BEK and does not modify any
+ * other properties of the bucket or the existing keys within it.
+ *
+ * Existing keys in the bucket will retain their current properties, and any
+ * keys added before the BEK reset will remain unencrypted. Keys added after the
+ * BEK reset will be encrypted using the new BEK details provided.
+ *
+ * @deprecated This functionality is deprecated as it is not intended for users
+ * to reset bucket encryption post-bucket creation under normal circumstances
+ * and may be removed in the future. Users are advised to exercise caution and
+ * consider alternative approaches for managing bucket encryption unless
+ * HDDS-7449 or HDDS-7526 is encountered. As a result, the setter methods and
+ * this CLI functionality have been marked as deprecated, and the command has
+ * been hidden.
+ */
+@Deprecated
+@CommandLine.Command(name = "set-encryption-key",
+    description = "Set Bucket Encryption Key (BEK) for a given bucket. Users " +
+        "are advised to exercise caution and consider alternative approaches " +
+        "for managing bucket encryption unless HDDS-7449 or HDDS-7526 is " +
+        "encountered.",
+    hidden = true)
+public class SetEncryptionKey extends BucketHandler {
+
+  @CommandLine.Option(names = {"--key", "-k"},
+      description = "bucket encryption key name")
+  private String bekName;
+
+  @Override
+  protected void execute(OzoneClient client, OzoneAddress address)
+      throws IOException, OzoneClientException {
+
+    String volumeName = address.getVolumeName();
+    String bucketName = address.getBucketName();
+    OzoneBucket bucket =
+        client.getObjectStore().getVolume(volumeName).getBucket(bucketName);
+    bucket.setEncryptionKey(bekName);
+  }
+}