HDDS-5244. Allow multiple OM request versions to be supported at same layout version.

.github/workflows/post-commit.yml

@@ -118,7 +118,7 @@ jobs:
   acceptance:
     needs: compile
     runs-on: ubuntu-18.04
-    timeout-minutes: 120
+    timeout-minutes: 120 # Todo HDDS-5270
     if: github.event_name != 'pull_request' || github.event.pull_request.draft == false
     strategy:
       matrix:
@@ -163,7 +163,7 @@ jobs:
         continue-on-error: true
   integration:
     runs-on: ubuntu-18.04
-    timeout-minutes: 100
+    timeout-minutes: 120
     if: github.event_name != 'pull_request' || github.event.pull_request.draft == false
     strategy:
       matrix:

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

@@ -48,6 +48,7 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus;
 import org.apache.hadoop.test.LambdaTestUtils;
 import org.junit.Assert;
+import org.junit.Ignore;
 import org.junit.Test;
 
 /**
@@ -74,28 +75,24 @@ public void setup() throws Exception {
   }
 
   /**
-   * Calls prepare on all OMs when they have no transaction information.
-   * Checks that they are brought into prepare mode successfully.
+   * Writes data to the cluster via the leader OM, and then prepares it.
+   * Checks that every OM is prepared successfully.
    */
   @Test
-  public void testPrepareWithoutTransactions() throws Exception {
+  public void testPrepareWithTransactions() throws Exception {
     setup();
+
     long prepareIndex = submitPrepareRequest();
     assertClusterPrepared(prepareIndex);
     assertRatisLogsCleared();
-  }
 
-  /**
-   * Writes data to the cluster via the leader OM, and then prepares it.
-   * Checks that every OM is prepared successfully.
-   */
-  @Test
-  public void testPrepareWithTransactions() throws Exception {
-    setup();
+    submitCancelPrepareRequest();
+    assertClusterNotPrepared();
+
     String volumeName = VOLUME + UUID.randomUUID().toString();
     Set<String> writtenKeys = writeKeysAndWaitForLogs(volumeName, 50,
         cluster.getOzoneManagersList());
-    long prepareIndex = submitPrepareRequest();
+    prepareIndex = submitPrepareRequest();
 
     // Make sure all OMs are prepared and all OMs still have their data.
     assertClusterPrepared(prepareIndex);
@@ -171,6 +168,7 @@ public void testPrepareDownedOM() throws Exception {
     }
   }
 
+  @Ignore("Flaky test tracked in HDDS-5109")
   @Test
   public void testPrepareWithRestart() throws Exception {
     setup();
@@ -188,6 +186,8 @@ public void testPrepareWithRestart() throws Exception {
     assertClusterPrepared(prepareIndex);
   }
 
+  @Ignore("Saving on CI time since this is a pessimistic test. We should not " +
+      "be able to do anything with 2 OMs down.")
   @Test
   public void testPrepareFailsWhenTwoOmsAreDown() throws Exception {
 

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

@@ -109,7 +109,7 @@ private void submitRequest(OzoneManagerProtocolProtos.OMRequest omRequest)
     ozoneManager.getMetrics().incNumTrashWriteRequests();
     if (ozoneManager.isRatisEnabled()) {
       OMClientRequest omClientRequest =
-          OzoneManagerRatisUtils.getRequest(ozoneManager, omRequest);
+          OzoneManagerRatisUtils.createClientRequest(omRequest);
       omRequest = omClientRequest.preExecute(ozoneManager);
       RaftClientRequest req = getRatisRequest(omRequest);
       ozoneManager.getOmRatisServer().submitRequest(omRequest, req);

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

@@ -18,23 +18,49 @@
 package org.apache.hadoop.ozone.om.ratis.utils;
 
 import com.google.common.base.Preconditions;
-
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.utils.HAUtils;
 import org.apache.hadoop.ozone.om.OzoneManager;
 import org.apache.hadoop.ozone.om.codec.OMDBDefinition;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.ozone.om.request.bucket.OMBucketCreateRequest;
+import org.apache.hadoop.ozone.om.request.bucket.OMBucketDeleteRequest;
+import org.apache.hadoop.ozone.om.request.bucket.OMBucketSetPropertyRequest;
 import org.apache.hadoop.ozone.om.request.OMClientRequest;
 import org.apache.hadoop.ozone.om.request.bucket.acl.OMBucketAddAclRequest;
 import org.apache.hadoop.ozone.om.request.bucket.acl.OMBucketRemoveAclRequest;
 import org.apache.hadoop.ozone.om.request.bucket.acl.OMBucketSetAclRequest;
+import org.apache.hadoop.ozone.om.request.file.OMDirectoryCreateRequest;
+import org.apache.hadoop.ozone.om.request.file.OMFileCreateRequest;
+import org.apache.hadoop.ozone.om.request.key.OMKeysDeleteRequest;
+import org.apache.hadoop.ozone.om.request.key.OMAllocateBlockRequest;
+import org.apache.hadoop.ozone.om.request.key.OMKeyCommitRequest;
+import org.apache.hadoop.ozone.om.request.key.OMKeyCreateRequest;
+import org.apache.hadoop.ozone.om.request.key.OMKeyDeleteRequest;
+import org.apache.hadoop.ozone.om.request.key.OMKeyPurgeRequest;
+import org.apache.hadoop.ozone.om.request.key.OMKeyRenameRequest;
+import org.apache.hadoop.ozone.om.request.key.OMKeysRenameRequest;
+import org.apache.hadoop.ozone.om.request.key.OMTrashRecoverRequest;
 import org.apache.hadoop.ozone.om.request.key.acl.OMKeyAddAclRequest;
 import org.apache.hadoop.ozone.om.request.key.acl.OMKeyRemoveAclRequest;
 import org.apache.hadoop.ozone.om.request.key.acl.OMKeySetAclRequest;
 import org.apache.hadoop.ozone.om.request.key.acl.prefix.OMPrefixAddAclRequest;
 import org.apache.hadoop.ozone.om.request.key.acl.prefix.OMPrefixRemoveAclRequest;
 import org.apache.hadoop.ozone.om.request.key.acl.prefix.OMPrefixSetAclRequest;
+import org.apache.hadoop.ozone.om.request.s3.multipart.S3InitiateMultipartUploadRequest;
+import org.apache.hadoop.ozone.om.request.s3.multipart.S3MultipartUploadAbortRequest;
+import org.apache.hadoop.ozone.om.request.s3.multipart.S3MultipartUploadCommitPartRequest;
+import org.apache.hadoop.ozone.om.request.s3.multipart.S3MultipartUploadCompleteRequest;
+import org.apache.hadoop.ozone.om.request.s3.security.S3GetSecretRequest;
+import org.apache.hadoop.ozone.om.request.security.OMCancelDelegationTokenRequest;
+import org.apache.hadoop.ozone.om.request.security.OMGetDelegationTokenRequest;
+import org.apache.hadoop.ozone.om.request.security.OMRenewDelegationTokenRequest;
+import org.apache.hadoop.ozone.om.request.upgrade.OMCancelPrepareRequest;
+import org.apache.hadoop.ozone.om.request.upgrade.OMFinalizeUpgradeRequest;
+import org.apache.hadoop.ozone.om.request.upgrade.OMPrepareRequest;
+import org.apache.hadoop.ozone.om.request.volume.OMVolumeCreateRequest;
+import org.apache.hadoop.ozone.om.request.volume.OMVolumeDeleteRequest;
 import org.apache.hadoop.ozone.om.request.volume.OMVolumeSetOwnerRequest;
 import org.apache.hadoop.ozone.om.request.volume.OMVolumeSetQuotaRequest;
 import org.apache.hadoop.ozone.om.request.volume.acl.OMVolumeAddAclRequest;
@@ -46,8 +72,6 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 import org.rocksdb.RocksDBException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.nio.file.Path;
@@ -58,110 +82,133 @@
  */
 public final class OzoneManagerRatisUtils {
 
-  private static final Logger LOG = LoggerFactory
-      .getLogger(OzoneManagerRatisUtils.class);
-
   private OzoneManagerRatisUtils() {
   }
-
-  public static OMClientRequest getRequest(OzoneManager om,
-                                           OMRequest omRequest) {
+  /**
+   * Create OMClientRequest which encapsulates the OMRequest.
+   * @param omRequest
+   * @return OMClientRequest
+   * @throws IOException
+   */
+  public static OMClientRequest createClientRequest(OMRequest omRequest) {
     Type cmdType = omRequest.getCmdType();
     switch (cmdType) {
+    case CreateVolume:
+      return new OMVolumeCreateRequest(omRequest);
+    case SetVolumeProperty:
+      boolean hasQuota = omRequest.getSetVolumePropertyRequest()
+          .hasQuotaInBytes();
+      boolean hasOwner = omRequest.getSetVolumePropertyRequest().hasOwnerName();
+      Preconditions.checkState(hasOwner || hasQuota, "Either Quota or owner " +
+          "should be set in the SetVolumeProperty request");
+      Preconditions.checkState(!(hasOwner && hasQuota), "Either Quota or " +
+          "owner should be set in the SetVolumeProperty request. Should not " +
+          "set both");
+      if (hasQuota) {
+        return new OMVolumeSetQuotaRequest(omRequest);
+      } else {
+        return new OMVolumeSetOwnerRequest(omRequest);
+      }
+    case DeleteVolume:
+      return new OMVolumeDeleteRequest(omRequest);
+    case CreateBucket:
+      return new OMBucketCreateRequest(omRequest);
+    case DeleteBucket:
+      return new OMBucketDeleteRequest(omRequest);
+    case SetBucketProperty:
+      return new OMBucketSetPropertyRequest(omRequest);
+    case AllocateBlock:
+      return new OMAllocateBlockRequest(omRequest);
+    case CreateKey:
+      return new OMKeyCreateRequest(omRequest);
+    case CommitKey:
+      return new OMKeyCommitRequest(omRequest);
+    case DeleteKey:
+      return new OMKeyDeleteRequest(omRequest);
+    case DeleteKeys:
+      return new OMKeysDeleteRequest(omRequest);
+    case RenameKey:
+      return new OMKeyRenameRequest(omRequest);
+    case RenameKeys:
+      return new OMKeysRenameRequest(omRequest);
+    case CreateDirectory:
+      return new OMDirectoryCreateRequest(omRequest);
+    case CreateFile:
+      return new OMFileCreateRequest(omRequest);
+    case PurgeKeys:
+      return new OMKeyPurgeRequest(omRequest);
+    case InitiateMultiPartUpload:
+      return new S3InitiateMultipartUploadRequest(omRequest);
+    case CommitMultiPartUpload:
+      return new S3MultipartUploadCommitPartRequest(omRequest);
+    case AbortMultiPartUpload:
+      return new S3MultipartUploadAbortRequest(omRequest);
+    case CompleteMultiPartUpload:
+      return new S3MultipartUploadCompleteRequest(omRequest);
     case AddAcl:
     case RemoveAcl:
     case SetAcl:
-      return getOMAclRequest(om, omRequest);
-    case SetVolumeProperty:
-      return getVolumeSetPropertyRequest(om, omRequest);
+      return getOMAclRequest(omRequest);
+    case GetDelegationToken:
+      return new OMGetDelegationTokenRequest(omRequest);
+    case CancelDelegationToken:
+      return new OMCancelDelegationTokenRequest(omRequest);
+    case RenewDelegationToken:
+      return new OMRenewDelegationTokenRequest(omRequest);
+    case GetS3Secret:
+      return new S3GetSecretRequest(omRequest);
+    case RecoverTrash:
+      return new OMTrashRecoverRequest(omRequest);
+    case FinalizeUpgrade:
+      return new OMFinalizeUpgradeRequest(omRequest);
+    case Prepare:
+      return new OMPrepareRequest(omRequest);
+    case CancelPrepare:
+      return new OMCancelPrepareRequest(omRequest);
     default:
-      Class<? extends OMClientRequest> requestClass =
-          om.getVersionManager()
-              .getHandler(omRequest.getCmdType().name());
-      return getClientRequest(requestClass, omRequest);
-    }
-  }
-
-  private static OMClientRequest getClientRequest(Class<?
-      extends OMClientRequest> requestClass, OMRequest omRequest) {
-    try {
-      return requestClass.getDeclaredConstructor(OMRequest.class)
-          .newInstance(omRequest);
-    } catch (Exception ex) {
-      LOG.error("Unable to get request handler for '{}', current layout " +
-              "version = {}, request factory returned '{}'",
-          omRequest.getCmdType(),
-          omRequest.getLayoutVersion().getVersion(),
-          requestClass.getSimpleName());
+      throw new IllegalStateException("Unrecognized write command " +
+          "type request" + cmdType);
     }
-    throw new IllegalStateException("Unrecognized write command " +
-        "type request : " + omRequest.getCmdType());
   }
 
-  public static OMClientRequest getOMAclRequest(OzoneManager om,
-                                                OMRequest omRequest) {
+  private static OMClientRequest getOMAclRequest(OMRequest omRequest) {
     Type cmdType = omRequest.getCmdType();
-    String requestType = null;
     if (Type.AddAcl == cmdType) {
       ObjectType type = omRequest.getAddAclRequest().getObj().getResType();
       if (ObjectType.VOLUME == type) {
-        requestType = OMVolumeAddAclRequest.getRequestType();
+        return new OMVolumeAddAclRequest(omRequest);
       } else if (ObjectType.BUCKET == type) {
-        requestType = OMBucketAddAclRequest.getRequestType();
+        return new OMBucketAddAclRequest(omRequest);
       } else if (ObjectType.KEY == type) {
-        requestType = OMKeyAddAclRequest.getRequestType();
+        return new OMKeyAddAclRequest(omRequest);
       } else {
-        requestType = OMPrefixAddAclRequest.getRequestType();
+        return new OMPrefixAddAclRequest(omRequest);
       }
     } else if (Type.RemoveAcl == cmdType) {
       ObjectType type = omRequest.getRemoveAclRequest().getObj().getResType();
       if (ObjectType.VOLUME == type) {
-        requestType = OMVolumeRemoveAclRequest.getRequestType();
+        return new OMVolumeRemoveAclRequest(omRequest);
       } else if (ObjectType.BUCKET == type) {
-        requestType = OMBucketRemoveAclRequest.getRequestType();
+        return new OMBucketRemoveAclRequest(omRequest);
       } else if (ObjectType.KEY == type) {
-        requestType = OMKeyRemoveAclRequest.getRequestType();
+        return new OMKeyRemoveAclRequest(omRequest);
       } else {
-        requestType = OMPrefixRemoveAclRequest.getRequestType();
+        return new OMPrefixRemoveAclRequest(omRequest);
       }
     } else {
       ObjectType type = omRequest.getSetAclRequest().getObj().getResType();
       if (ObjectType.VOLUME == type) {
-        requestType = OMVolumeSetAclRequest.getRequestType();
+        return new OMVolumeSetAclRequest(omRequest);
       } else if (ObjectType.BUCKET == type) {
-        requestType = OMBucketSetAclRequest.getRequestType();
+        return new OMBucketSetAclRequest(omRequest);
       } else if (ObjectType.KEY == type) {
-        requestType = OMKeySetAclRequest.getRequestType();
+        return new OMKeySetAclRequest(omRequest);
       } else {
-        requestType = OMPrefixSetAclRequest.getRequestType();
+        return new OMPrefixSetAclRequest(omRequest);
       }
     }
-    Class<? extends OMClientRequest> requestClass =
-        om.getVersionManager().getHandler(requestType);
-    return getClientRequest(requestClass, omRequest);
   }
 
-  public static OMClientRequest getVolumeSetPropertyRequest(
-      OzoneManager om, OMRequest omRequest) {
-    boolean hasQuota = omRequest.getSetVolumePropertyRequest()
-        .hasQuotaInBytes();
-    boolean hasOwner = omRequest.getSetVolumePropertyRequest().hasOwnerName();
-    Preconditions.checkState(hasOwner || hasQuota,
-        "Either Quota or owner " +
-            "should be set in the SetVolumeProperty request");
-    Preconditions.checkState(!(hasOwner && hasQuota),
-        "Either Quota or " +
-            "owner should be set in the SetVolumeProperty request. Should not "
-            + "set both");
-
-    String requestType = hasQuota ? OMVolumeSetQuotaRequest.getRequestType() :
-        OMVolumeSetOwnerRequest.getRequestType();
-    Class<? extends OMClientRequest> requestClass =
-        om.getVersionManager().getHandler(requestType);
-    return getClientRequest(requestClass, omRequest);
-  }
-
-
   /**
    * Convert exception result to {@link OzoneManagerProtocolProtos.Status}.
    * @param exception