diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
index e6a85566fe2b..1420e1cfb070 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
@@ -290,11 +290,13 @@ public void close() throws IOException {
     transport.close();
   }
 
+
   /**
    * Returns a OMRequest builder with specified type.
    * @param cmdType type of the request
    */
-  private OMRequest.Builder createOMRequest(Type cmdType) {
+  @VisibleForTesting
+  public OMRequest.Builder createOMRequest(Type cmdType) {
     return OMRequest.newBuilder()
         .setCmdType(cmdType)
         .setVersion(ClientVersion.CURRENT_VERSION)
@@ -307,7 +309,8 @@ private OMRequest.Builder createOMRequest(Type cmdType) {
    * @return response from OM
    * @throws IOException thrown if any Protobuf service exception occurs
    */
-  private OMResponse submitRequest(OMRequest omRequest)
+  @VisibleForTesting
+  public OMResponse submitRequest(OMRequest omRequest)
       throws IOException {
     OMRequest.Builder  builder = OMRequest.newBuilder(omRequest);
     // Insert S3 Authentication information for each request.
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestBucketLayoutWithOlderClient.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestBucketLayoutWithAllClients.java
similarity index 53%
rename from hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestBucketLayoutWithOlderClient.java
rename to hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestBucketLayoutWithAllClients.java
index 73596781cc64..368d06711d6d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestBucketLayoutWithOlderClient.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestBucketLayoutWithAllClients.java
@@ -16,6 +16,7 @@
  */
 package org.apache.hadoop.ozone.om;
 
+import com.google.protobuf.ServiceException;
 import org.apache.hadoop.hdds.utils.IOUtils;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.ClientVersion;
@@ -23,29 +24,39 @@
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.ozone.client.OzoneBucket;
 import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolClientSideTranslatorPB;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.UUID;
+import java.util.function.Function;
+import java.util.stream.Stream;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 /**
- * Tests to verify bucket ops with older version client.
+ * Tests to verify bucket ops with different client versions.
  */
 @Timeout(1200)
-public class TestBucketLayoutWithOlderClient {
+public class TestBucketLayoutWithAllClients {
 
   private static MiniOzoneCluster cluster = null;
   private static OzoneConfiguration conf;
   private static OzoneClient client;
+  private static OzoneManagerProtocolClientSideTranslatorPB ozoneManagerClient;
 
   /**
    * Create a MiniDFSCluster for testing.
@@ -61,6 +72,21 @@ public static void init() throws Exception {
     cluster = MiniOzoneCluster.newBuilder(conf).build();
     cluster.waitForClusterToBeReady();
     client = cluster.newClient();
+    ozoneManagerClient =
+        (OzoneManagerProtocolClientSideTranslatorPB)
+            ((RpcClient) client.getObjectStore().getClientProxy()).getOzoneManagerClient();
+  }
+
+  private static Stream<Arguments> allClientVersions() {
+    return Arrays.stream(ClientVersion.values()).flatMap(
+        clientVersion -> Stream.of(Arguments.of(clientVersion, BucketLayout.FILE_SYSTEM_OPTIMIZED),
+            Arguments.of(clientVersion, BucketLayout.OBJECT_STORE)));
+  }
+
+  private OzoneManagerProtocolProtos.OMRequest.Builder getOmRequest(OzoneManagerProtocolProtos.Type type,
+                                                                    ClientVersion clientVersion) {
+    return ozoneManagerClient.createOMRequest(type)
+        .setVersion(clientVersion.toProtoValue());
   }
 
   @Test
@@ -109,6 +135,57 @@ public void testCreateBucketWithOlderClient() throws Exception {
     assertEquals(BucketLayout.LEGACY, bucketInfo.getBucketLayout());
   }
 
+  private OzoneManagerProtocolProtos.KeyArgs getKeyArgs(String volume, String bucket) {
+    return OzoneManagerProtocolProtos.KeyArgs.newBuilder()
+            .setVolumeName(volume)
+            .setBucketName(bucket).setKeyName("a/b/c/key")
+            .build();
+  }
+
+  private OzoneManagerProtocolProtos.Status createKey(String volume, String bucket, ClientVersion clientVersion,
+                            OzoneManagerProtocolProtos.Status expectedStatus) throws IOException {
+    OzoneManagerProtocolProtos.CreateFileRequest createFileRequest =
+        OzoneManagerProtocolProtos.CreateFileRequest.newBuilder()
+        .setKeyArgs(getKeyArgs(volume, bucket))
+        .setIsOverwrite(true)
+        .setIsRecursive(true).build();
+    OzoneManagerProtocolProtos.OMRequest request = getOmRequest(OzoneManagerProtocolProtos.Type.CreateFile,
+        clientVersion)
+        .setCreateFileRequest(createFileRequest).build();
+    OzoneManagerProtocolProtos.OMResponse
+    omResponse = ozoneManagerClient.submitRequest(request);
+    return expectedStatus;
+  }
+
+  private void performOpWithLatestClientVersion(ClientVersion clientVersion,
+                                                Function<ClientVersion, OzoneManagerProtocolProtos.Status> operation,
+                                                OzoneManagerProtocolProtos.Status expectedStatusWithClientVersion) {
+    Assertions.assertEquals(expectedStatusWithClientVersion, operation.apply(clientVersion));
+    if (!operation.apply(clientVersion).equals(OzoneManagerProtocolProtos.Status.OK)) {
+      operation.apply(ClientVersion.CURRENT);
+    }
+
+  }
+  @ParameterizedTest
+  @MethodSource("allClientVersions")
+  public void testBucketOperationsOnNonLegacyBucket(ClientVersion version, BucketLayout bucketLayout)
+      throws IOException, ServiceException {
+    String suffix = version.toProtoValue() + "-" +
+        bucketLayout.toString().toLowerCase().replaceAll("_","-");
+    String volume = "volume-" + suffix;
+    String bucket = "bucket-" + suffix;
+    OzoneBucket ozoneBucket = TestDataUtil.createVolumeAndBucket(client, volume, bucket, bucketLayout);
+    Assertions.assertEquals(bucketLayout,
+        client.getObjectStore().getVolume(volume).getBucket(bucket).getBucketLayout());
+    // Create key
+    OzoneManagerProtocolProtos.Status expectedStatus =
+        version.toProtoValue() >= ClientVersion.BUCKET_LAYOUT_SUPPORT.toProtoValue() ?
+            OzoneManagerProtocolProtos.Status.OK : OzoneManagerProtocolProtos.Status.NOT_SUPPORTED_OPERATION;
+    performOpWithLatestClientVersion(createKey(volume, bucket, version, expectedStatus));
+
+
+  }
+
   /**
    * Shutdown MiniDFSCluster.
    */
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketDeleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketDeleteRequest.java
index 48a3c807b9a3..0d828f14cc01 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketDeleteRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketDeleteRequest.java
@@ -23,6 +23,7 @@
 import java.util.Iterator;
 import java.util.Map;
 
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.TableIterator;
@@ -287,6 +288,10 @@ private boolean bucketContainsSnapshotInCache(
   )
   public static OMRequest blockBucketDeleteWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     DeleteBucketRequest request = req.getDeleteBucketRequest();
 
     if (request.hasBucketName() && request.hasVolumeName()) {
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMDirectoryCreateRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMDirectoryCreateRequest.java
index 6071110a315d..a00cc0b29e1c 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMDirectoryCreateRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMDirectoryCreateRequest.java
@@ -31,6 +31,7 @@
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
@@ -438,6 +439,10 @@ public static OMRequest disallowCreateDirectoryWithECReplicationConfig(
   )
   public static OMRequest blockCreateDirectoryWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getCreateDirectoryRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getCreateDirectoryRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java
index 9b9fb4e7cc5c..83bdda6c04a7 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java
@@ -29,6 +29,7 @@
 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -431,6 +432,10 @@ public static OMRequest disallowCreateFileWithECReplicationConfig(
   )
   public static OMRequest blockCreateFileWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getCreateFileRequest().hasKeyArgs()) {
 
       KeyArgs keyArgs = req.getCreateFileRequest().getKeyArgs();
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java
index ac4d9ab6244f..564f48c83e95 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java
@@ -26,6 +26,7 @@
 
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
@@ -304,6 +305,11 @@ public static OMRequest disallowAllocateBlockWithECReplicationConfig(
   )
   public static OMRequest blockAllocateBlockWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
+
     if (req.getAllocateBlockRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getAllocateBlockRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java
index 7d31422c80e3..33afb1db07f7 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java
@@ -28,6 +28,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
@@ -469,6 +470,10 @@ public static OMRequest disallowCommitKeyWithECReplicationConfig(
   )
   public static OMRequest blockCommitKeyWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getCommitKeyRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getCommitKeyRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java
index 36f2a8c31be1..017407c3d88b 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java
@@ -28,6 +28,7 @@
 
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.om.OMConfigKeys;
@@ -430,6 +431,10 @@ public static OMRequest disallowCreateKeyWithECReplicationConfig(
   )
   public static OMRequest blockCreateKeyWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getCreateKeyRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getCreateKeyRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyDeleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyDeleteRequest.java
index 61e5976f8052..24f5c39c04d1 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyDeleteRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyDeleteRequest.java
@@ -23,6 +23,7 @@
 import java.util.Map;
 
 import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OmUtils;
@@ -239,6 +240,10 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
   )
   public static OMRequest blockDeleteKeyWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getDeleteKeyRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getDeleteKeyRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRenameRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRenameRequest.java
index db08951831cb..56515c54d532 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRenameRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRenameRequest.java
@@ -23,6 +23,7 @@
 import java.util.Map;
 
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -276,6 +277,10 @@ private Map<String, String> buildAuditMap(
   )
   public static OMRequest blockRenameKeyWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getRenameKeyRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getRenameKeyRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysDeleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysDeleteRequest.java
index 19e9ed716e4a..0e95119f84f1 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysDeleteRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysDeleteRequest.java
@@ -20,6 +20,7 @@
 
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
@@ -357,6 +358,10 @@ protected static void addDeletedKeys(Map<String, String> auditMap,
   )
   public static OMRequest blockDeleteKeysWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getDeleteKeysRequest().hasDeleteKeys()) {
       DeleteKeyArgs keyArgs = req.getDeleteKeysRequest().getDeleteKeys();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysRenameRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysRenameRequest.java
index 0b93dd57946e..8ea13d1a721e 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysRenameRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysRenameRequest.java
@@ -19,6 +19,7 @@
 package org.apache.hadoop.ozone.om.request.key;
 
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
@@ -303,6 +304,10 @@ private Map<String, String> buildAuditMap(Map<String, String> auditMap,
   )
   public static OMRequest blockRenameKeysWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getRenameKeysRequest().hasRenameKeysArgs()) {
       RenameKeysArgs keyArgs = req.getRenameKeysRequest().getRenameKeysArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAddAclRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAddAclRequest.java
index 31fa245fa55b..8a111b078a79 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAddAclRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAddAclRequest.java
@@ -23,6 +23,7 @@
 import java.util.Map;
 
 import com.google.common.collect.Lists;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -171,6 +172,10 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
   )
   public static OMRequest blockAddAclWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getAddAclRequest().hasObj()) {
       OzoneObj obj = OzoneObjInfo.fromProtobuf(req.getAddAclRequest().getObj());
       String path = obj.getPath();
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyRemoveAclRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyRemoveAclRequest.java
index 7e4fc04eb39c..634152cc3a9f 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyRemoveAclRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyRemoveAclRequest.java
@@ -23,6 +23,7 @@
 import java.util.Map;
 
 import com.google.common.collect.Lists;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -172,6 +173,10 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
   )
   public static OMRequest blockRemoveAclWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getRemoveAclRequest().hasObj()) {
       OzoneObj obj =
           OzoneObjInfo.fromProtobuf(req.getRemoveAclRequest().getObj());
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeySetAclRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeySetAclRequest.java
index 0e5e1cf3362a..c6f7f015a1cf 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeySetAclRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeySetAclRequest.java
@@ -23,6 +23,7 @@
 import java.util.Map;
 
 import com.google.common.collect.Lists;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -168,6 +169,10 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
   )
   public static OMRequest blockSetAclWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getSetAclRequest().hasObj()) {
       OzoneObj obj =
           OzoneObjInfo.fromProtobuf(req.getSetAclRequest().getObj());
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java
index f16ef9f8f423..269adea42b55 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java
@@ -20,6 +20,7 @@
 
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.om.helpers.KeyValueUtil;
 import org.apache.ratis.server.protocol.TermIndex;
@@ -326,6 +327,10 @@ protected void logResult(OzoneManager ozoneManager,
   )
   public static OMRequest blockInitiateMPUWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getInitiateMultiPartUploadRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getInitiateMultiPartUploadRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java
index 400291d7a448..f3b614a3bb13 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java
@@ -22,6 +22,7 @@
 import java.nio.file.InvalidPathException;
 import java.util.Map;
 
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
@@ -295,6 +296,10 @@ public static OMRequest disallowAbortMultiPartUploadWithECReplicationConfig(
   )
   public static OMRequest blockMPUAbortWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getAbortMultiPartUploadRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getAbortMultiPartUploadRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java
index 1123b8dba7a6..01a990a6245f 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java
@@ -19,6 +19,7 @@
 package org.apache.hadoop.ozone.om.request.s3.multipart;
 
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.audit.OMAction;
@@ -386,6 +387,10 @@ public static OMRequest disallowCommitMultiPartUploadWithECReplicationConfig(
   )
   public static OMRequest blockMPUCommitWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getCommitMultiPartUploadRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getCommitMultiPartUploadRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java
index 1dab110e783a..e8b5b18e14af 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java
@@ -32,6 +32,7 @@
 import java.util.function.BiFunction;
 
 import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.om.OzoneConfigUtil;
 import org.apache.hadoop.ozone.om.request.file.OMDirectoryCreateRequestWithFSO;
 import org.apache.hadoop.ozone.om.request.file.OMFileRequest;
@@ -748,6 +749,10 @@ private void updateCache(OMMetadataManager omMetadataManager,
   )
   public static OMRequest blockMPUCompleteWithBucketLayoutFromOldClient(
       OMRequest req, ValidationContext ctx) throws IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return req;
+    }
     if (req.getCompleteMultiPartUploadRequest().hasKeyArgs()) {
       KeyArgs keyArgs = req.getCompleteMultiPartUploadRequest().getKeyArgs();
 
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
index 5b8365e5a75b..5c4387db9a16 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
@@ -40,6 +40,7 @@
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.scm.protocolPB.OzonePBHelper;
 import org.apache.hadoop.hdds.utils.FaultInjector;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.util.PayloadUtils;
 import org.apache.hadoop.ozone.om.OzoneManager;
@@ -636,6 +637,9 @@ private GetKeyInfoResponse getKeyInfo(GetKeyInfoRequest request,
   public static OMResponse disallowLookupKeyResponseWithECReplicationConfig(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.ERASURE_CODING_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasLookupKeyResponse()) {
       return resp;
     }
@@ -663,7 +667,11 @@ public static OMResponse disallowLookupKeyResponseWithECReplicationConfig(
   public static OMResponse disallowLookupKeyWithBucketLayout(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException, IOException {
-    if (!resp.hasLookupKeyResponse()) {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return resp;
+    }
+    if (ClientVersion.fromProtoValue(req.getVersion())
+        .compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0 || !resp.hasLookupKeyResponse()) {
       return resp;
     }
     KeyInfo keyInfo = resp.getLookupKeyResponse().getKeyInfo();
@@ -750,6 +758,9 @@ private ListKeysLightResponse listKeysLight(ListKeysRequest request)
   public static OMResponse disallowListKeysResponseWithECReplicationConfig(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.ERASURE_CODING_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasListKeysResponse()) {
       return resp;
     }
@@ -777,6 +788,9 @@ public static OMResponse disallowListKeysResponseWithECReplicationConfig(
   public static OMResponse disallowListKeysWithBucketLayout(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException, IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasListKeysResponse()) {
       return resp;
     }
@@ -838,6 +852,9 @@ private ListTrashResponse listTrash(ListTrashRequest request,
   public static OMResponse disallowListTrashWithECReplicationConfig(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.ERASURE_CODING_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasListTrashResponse()) {
       return resp;
     }
@@ -868,6 +885,9 @@ public static OMResponse disallowListTrashWithECReplicationConfig(
   public static OMResponse disallowListTrashWithBucketLayout(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException, IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasListTrashResponse()) {
       return resp;
     }
@@ -1055,6 +1075,9 @@ private RefetchSecretKeyResponse refetchSecretKey() {
   public static OMResponse disallowGetFileStatusWithECReplicationConfig(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.ERASURE_CODING_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasGetFileStatusResponse()) {
       return resp;
     }
@@ -1085,6 +1108,9 @@ public static OMResponse disallowGetFileStatusWithECReplicationConfig(
   public static OMResponse disallowGetFileStatusWithBucketLayout(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException, IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasGetFileStatusResponse()) {
       return resp;
     }
@@ -1134,6 +1160,9 @@ private LookupFileResponse lookupFile(LookupFileRequest request,
   public static OMResponse disallowLookupFileWithECReplicationConfig(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.ERASURE_CODING_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasLookupFileResponse()) {
       return resp;
     }
@@ -1162,6 +1191,9 @@ public static OMResponse disallowLookupFileWithECReplicationConfig(
   public static OMResponse disallowLookupFileWithBucketLayout(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException, IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasLookupFileResponse()) {
       return resp;
     }
@@ -1244,6 +1276,9 @@ private ListStatusLightResponse listStatusLight(
   public static OMResponse disallowListStatusResponseWithECReplicationConfig(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.ERASURE_CODING_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasListStatusResponse()) {
       return resp;
     }
@@ -1272,6 +1307,9 @@ public static OMResponse disallowListStatusResponseWithECReplicationConfig(
   public static OMResponse disallowListStatusResponseWithBucketLayout(
       OMRequest req, OMResponse resp, ValidationContext ctx)
       throws ServiceException, IOException {
+    if (ClientVersion.fromProtoValue(req.getVersion()).compareTo(ClientVersion.BUCKET_LAYOUT_SUPPORT) >= 0) {
+      return resp;
+    }
     if (!resp.hasListStatusResponse()) {
       return resp;
     }