HDDS-12161. Remove code paths for non-Ratis OM in request/response

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -511,7 +511,7 @@ public static File createOMDir(String dirPath) {
    * @return {@link RepeatedOmKeyInfo}
    */
   public static RepeatedOmKeyInfo prepareKeyForDelete(OmKeyInfo keyInfo,
-      long trxnLogIndex, boolean isRatisEnabled) {
+      long trxnLogIndex) {
     // If this key is in a GDPR enforced bucket, then before moving
     // KeyInfo to deletedTable, remove the GDPR related metadata and
     // FileEncryptionInfo from KeyInfo.
@@ -525,7 +525,7 @@ public static RepeatedOmKeyInfo prepareKeyForDelete(OmKeyInfo keyInfo,
     }
 
     // Set the updateID
-    keyInfo.setUpdateID(trxnLogIndex, isRatisEnabled);
+    keyInfo.setUpdateID(trxnLogIndex);
 
     //The key doesn't exist in deletedTable, so create a new instance.
     return new RepeatedOmKeyInfo(keyInfo);

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

@@ -174,15 +174,14 @@ private OmMultipartKeyInfo(Builder b) {
 
   /** Copy constructor. */
   private OmMultipartKeyInfo(OmMultipartKeyInfo b) {
+    super(b);
     this.uploadID = b.uploadID;
     this.creationTime = b.creationTime;
     this.replicationConfig = b.replicationConfig;
     // PartKeyInfoMap is an immutable data structure. Whenever a PartKeyInfo
     // is added, it returns a new shallow copy of the PartKeyInfoMap Object
     // so here we can directly pass in partKeyInfoMap
     this.partKeyInfoMap = b.partKeyInfoMap;
-    setObjectID(b.getObjectID());
-    setUpdateID(b.getUpdateID());
     this.parentID = b.parentID;
   }
 

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

@@ -35,6 +35,10 @@ protected WithMetadata(Builder b) {
     metadata = b.metadata;
   }
 
+  protected WithMetadata(WithMetadata other) {
+    metadata = new ConcurrentHashMap<>(other.getMetadata());
+  }
+
   /**
    * Custom key value metadata.
    */

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

@@ -37,6 +37,12 @@ protected WithObjectID(Builder b) {
     updateID = b.updateID;
   }
 
+  protected WithObjectID(WithObjectID other) {
+    super(other);
+    objectID = other.objectID;
+    updateID = other.updateID;
+  }
+
   /**
    * ObjectIDs are unique and immutable identifier for each object in the
    * System.
@@ -73,10 +79,8 @@ public final void setObjectID(long obId) {
   /**
    * Sets the update ID. For each modification of this object, we will set
    * this to a value greater than the current value.
-   * @param updateId  long
-   * @param isRatisEnabled boolean
    */
-  public final void setUpdateID(long updateId, boolean isRatisEnabled) {
+  public final void setUpdateID(long newValue) {
 
     // Because in non-HA, we have multiple rpc handler threads and
     // transactionID is generated in OzoneManagerServerSideTranslatorPB.
@@ -103,25 +107,22 @@ public final void setUpdateID(long updateId, boolean isRatisEnabled) {
     // Main reason, in non-HA transaction Index after restart starts from 0.
     // And also because of this same reason we don't do replay checks in non-HA.
 
-    if (isRatisEnabled && updateId < this.getUpdateID()) {
+    final long currentValue = getUpdateID();
+    if (newValue < currentValue) {
       throw new IllegalArgumentException(String.format(
           "Trying to set updateID to %d which is not greater than the " +
-              "current value of %d for %s", updateId, this.getUpdateID(),
+              "current value of %d for %s", newValue, currentValue,
           getObjectInfo()));
     }
 
-    this.setUpdateID(updateId);
+    updateID = newValue;
   }
 
   /** Hook method, customized in subclasses. */
   public String getObjectInfo() {
     return this.toString();
   }
 
-  public final void setUpdateID(long updateID) {
-    this.updateID = updateID;
-  }
-
   /** Builder for {@link WithObjectID}. */
   public static class Builder extends WithMetadata.Builder {
     private long objectID;

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

@@ -179,7 +179,6 @@
 import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.WRITE;
 import static org.apache.ozone.test.GenericTestUtils.getTestStartTime;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assumptions.assumeThat;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -4934,8 +4933,6 @@ public void testUploadWithStreamAndMemoryMappedBuffer(@TempDir Path dir) throws
   @Test
   public void testParallelDeleteBucketAndCreateKey() throws IOException,
       InterruptedException, TimeoutException {
-    assumeThat(getCluster().getOzoneManager().isRatisEnabled()).isTrue();
-
     String volumeName = UUID.randomUUID().toString();
     String bucketName = UUID.randomUUID().toString();
 

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

@@ -193,7 +193,6 @@ public void setUp() throws IOException {
     omMetadataManager = new OmMetadataManagerImpl(conf, ozoneManager);
     when(ozoneManager.getMetrics()).thenReturn(omMetrics);
     when(ozoneManager.getMetadataManager()).thenReturn(omMetadataManager);
-    when(ozoneManager.isRatisEnabled()).thenReturn(true);
     auditLogger = mock(AuditLogger.class);
     when(ozoneManager.getAuditLogger()).thenReturn(auditLogger);
     doNothing().when(auditLogger).logWrite(any(AuditMessage.class));

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

@@ -4233,14 +4233,6 @@ public void checkLeaderStatus() throws OMNotLeaderException,
     }
   }
 
-  /**
-   * Return if Ratis is enabled or not.
-   */
-  // TODO remove in HDDS-12161
-  public boolean isRatisEnabled() {
-    return true;
-  }
-
   /**
    * @return true if Ozone filesystem snapshot is enabled, false otherwise.
    */

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

@@ -99,7 +99,6 @@ OMClientResponse getResponse() {
   public static final class Builder {
     private OMMetadataManager omMetadataManager;
     private Consumer<TermIndex> updateLastAppliedIndex = termIndex -> { };
-    private boolean isRatisEnabled = false;
     private boolean isTracingEnabled = false;
     private int maxUnFlushedTransactionCount = 0;
     private FlushNotifier flushNotifier;
@@ -118,11 +117,6 @@ Builder setUpdateLastAppliedIndex(Consumer<TermIndex> updateLastAppliedIndex) {
       return this;
     }
 
-    public Builder enableRatis(boolean enableRatis) {
-      this.isRatisEnabled = enableRatis;
-      return this;
-    }
-
     public Builder enableTracing(boolean enableTracing) {
       this.isTracingEnabled = enableTracing;
       return this;
@@ -149,9 +143,8 @@ public Builder setS3SecretManager(S3SecretManager s3SecretManager) {
     }
 
     public OzoneManagerDoubleBuffer build() {
-      Preconditions.assertTrue(isRatisEnabled == maxUnFlushedTransactionCount > 0L,
-          () -> "Ratis is " + (isRatisEnabled ? "enabled" : "disabled")
-              + " but maxUnFlushedTransactionCount = " + maxUnFlushedTransactionCount);
+      Preconditions.assertTrue(maxUnFlushedTransactionCount > 0L,
+          () -> "maxUnFlushedTransactionCount = " + maxUnFlushedTransactionCount);
       if (flushNotifier == null) {
         flushNotifier = new FlushNotifier();
       }
@@ -172,7 +165,6 @@ static Semaphore newSemaphore(int permits) {
   private Queue<Entry> readyBuffer;
   /**
    * Limit the number of un-flushed transactions for {@link OzoneManagerStateMachine}.
-   * It is set to null if ratis is disabled; see {@link #isRatisEnabled()}.
    */
   private final Semaphore unFlushedTransactions;
 
@@ -221,16 +213,11 @@ public OzoneManagerDoubleBuffer start() {
     return this;
   }
 
-  private boolean isRatisEnabled() {
-    return unFlushedTransactions != null;
-  }
-
   /**
    * Acquires the given number of permits from unFlushedTransactions,
    * blocking until all are available, or the thread is interrupted.
    */
   public void acquireUnFlushedTransactions(int n) throws InterruptedException {
-    Preconditions.assertTrue(isRatisEnabled(), "Ratis is not enabled");
     unFlushedTransactions.acquire(n);
   }
 
@@ -369,15 +356,6 @@ private void flushBatch(Queue<Entry> buffer) throws IOException {
       metrics.updateFlushTime(Time.monotonicNow() - startTime);
     }
 
-    // Complete futures first and then do other things.
-    // So that handler threads will be released.
-    if (!isRatisEnabled()) {
-      buffer.stream()
-          .map(Entry::getResponse)
-          .map(OMClientResponse::getFlushFuture)
-          .forEach(f -> f.complete(null));
-    }
-
     final long accumulativeCount = flushedTransactionCount.addAndGet(flushedTransactionsSize);
     final long flushedIterations = flushIterations.incrementAndGet();
     LOG.debug("Sync iteration: {}, size in this iteration: {}, accumulative count: {}",
@@ -386,9 +364,7 @@ private void flushBatch(Queue<Entry> buffer) throws IOException {
     // Clean up committed transactions.
     cleanupCache(cleanupEpochs);
 
-    if (isRatisEnabled()) {
-      releaseUnFlushedTransactions(flushedTransactionsSize);
-    }
+    releaseUnFlushedTransactions(flushedTransactionsSize);
     // update the last updated index in OzoneManagerStateMachine.
     updateLastAppliedIndex.accept(lastTransaction);
 
@@ -555,10 +531,6 @@ private void terminate(Throwable t, int status, OMResponse omResponse) {
   public synchronized void add(OMClientResponse response, TermIndex termIndex) {
     currentBuffer.add(new Entry(termIndex, response));
     notify();
-
-    if (!isRatisEnabled()) {
-      response.setFlushFuture(new CompletableFuture<>());
-    }
   }
 
   /**

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

@@ -463,7 +463,6 @@ public OzoneManagerDoubleBuffer buildDoubleBufferForRatis() {
         .setMaxUnFlushedTransactionCount(maxUnFlushedTransactionCount)
         .setThreadPrefix(threadPrefix)
         .setS3SecretManager(ozoneManager.getS3SecretManager())
-        .enableRatis(true)
         .enableTracing(isTracingEnabled)
         .build()
         .start();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketCreateRequest.java

@@ -245,8 +245,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       // Add objectID and updateID
       omBucketInfo.setObjectID(
           ozoneManager.getObjectIdFromTxId(transactionLogIndex));
-      omBucketInfo.setUpdateID(transactionLogIndex,
-          ozoneManager.isRatisEnabled());
+      omBucketInfo.setUpdateID(transactionLogIndex);
 
       addDefaultAcls(omBucketInfo, omVolumeArgs, ozoneManager);
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketSetOwnerRequest.java

@@ -155,8 +155,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       omBucketInfo.setModificationTime(
           setBucketPropertyRequest.getModificationTime());
       // Set the updateID to current transaction log index
-      omBucketInfo.setUpdateID(transactionLogIndex,
-          ozoneManager.isRatisEnabled());
+      omBucketInfo.setUpdateID(transactionLogIndex);
 
       // Update table cache.
       omMetadataManager.getBucketTable().addCacheEntry(

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/acl/OMBucketAclRequest.java

@@ -108,8 +108,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       }
 
       operationResult = omBucketAclOp.test(ozoneAcls, omBucketInfo);
-      omBucketInfo.setUpdateID(transactionLogIndex,
-          ozoneManager.isRatisEnabled());
+      omBucketInfo.setUpdateID(transactionLogIndex);
 
       if (operationResult) {
         // Update the modification time when updating ACLs of Bucket.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java

@@ -247,7 +247,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           keyArgs.getDataSize(), locations, getFileEncryptionInfo(keyArgs),
           ozoneManager.getPrefixManager(), omBucketInfo, pathInfo, trxnLogIndex,
           ozoneManager.getObjectIdFromTxId(trxnLogIndex),
-          ozoneManager.isRatisEnabled(), repConfig, ozoneManager.getConfiguration());
+          repConfig, ozoneManager.getConfiguration());
       validateEncryptionKeyInfo(omBucketInfo, keyArgs);
 
       long openVersion = omKeyInfo.getLatestVersionLocations().getVersion();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequestWithFSO.java

@@ -170,7 +170,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
               getFileEncryptionInfo(keyArgs), ozoneManager.getPrefixManager(),
               bucketInfo, pathInfoFSO, trxnLogIndex,
               pathInfoFSO.getLeafNodeObjectId(),
-              ozoneManager.isRatisEnabled(), repConfig, ozoneManager.getConfiguration());
+          repConfig, ozoneManager.getConfiguration());
       validateEncryptionKeyInfo(bucketInfo, keyArgs);
 
       long openVersion = omFileInfo.getLatestVersionLocations().getVersion();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMRecoverLeaseRequest.java

@@ -238,7 +238,7 @@ private RecoverLeaseResponse doWork(OzoneManager ozoneManager,
             KEY_UNDER_LEASE_SOFT_LIMIT_PERIOD);
       }
       openKeyInfo.getMetadata().put(OzoneConsts.LEASE_RECOVERY, "true");
-      openKeyInfo.setUpdateID(transactionLogIndex, ozoneManager.isRatisEnabled());
+      openKeyInfo.setUpdateID(transactionLogIndex);
       openKeyInfo.setModificationTime(Time.now());
       // add to cache.
       omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java

@@ -239,7 +239,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       openKeyInfo.setModificationTime(keyArgs.getModificationTime());
 
       // Set the UpdateID to current transactionLogIndex
-      openKeyInfo.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());
+      openKeyInfo.setUpdateID(trxnLogIndex);
 
       // Add to cache.
       omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequestWithFSO.java

@@ -162,7 +162,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       openKeyInfo.setModificationTime(keyArgs.getModificationTime());
 
       // Set the UpdateID to current transactionLogIndex
-      openKeyInfo.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());
+      openKeyInfo.setUpdateID(trxnLogIndex);
 
       // Add to cache.
       addOpenTableCacheEntry(trxnLogIndex, omMetadataManager, openKeyName, keyName,

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMDirectoriesPurgeRequestWithFSO.java

@@ -203,7 +203,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
     }
 
     return new OMDirectoriesPurgeResponseWithFSO(
-        omResponse.build(), purgeRequests, ozoneManager.isRatisEnabled(),
-            getBucketLayout(), volBucketInfoMap, fromSnapshotInfo, openKeyInfoMap);
+        omResponse.build(), purgeRequests,
+        getBucketLayout(), volBucketInfoMap, fromSnapshotInfo, openKeyInfoMap);
   }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java

@@ -278,7 +278,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         openKeyToDelete = omMetadataManager.getOpenKeyTable(getBucketLayout()).get(dbOpenKeyToDeleteKey);
         openKeyToDelete.getMetadata().put(OzoneConsts.OVERWRITTEN_HSYNC_KEY, "true");
         openKeyToDelete.setModificationTime(Time.now());
-        openKeyToDelete.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());
+        openKeyToDelete.setUpdateID(trxnLogIndex);
         omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
             dbOpenKeyToDeleteKey, openKeyToDelete, trxnLogIndex);
       }
@@ -310,7 +310,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           omKeyInfo.updateLocationInfoList(locationInfoList, false);
 
       // Set the UpdateID to current transactionLogIndex
-      omKeyInfo.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());
+      omKeyInfo.setUpdateID(trxnLogIndex);
 
       Map<String, RepeatedOmKeyInfo> oldKeyVersionsToDeleteMap = null;
       long correctedSpace = omKeyInfo.getReplicatedSize();
@@ -324,7 +324,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         // Subtract the size of blocks to be overwritten.
         correctedSpace -= keyToDelete.getReplicatedSize();
         RepeatedOmKeyInfo oldVerKeyInfo = getOldVersionsToCleanUp(
-            keyToDelete, trxnLogIndex, ozoneManager.isRatisEnabled());
+            keyToDelete, trxnLogIndex);
         checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
             correctedSpace);
         // using pseudoObjId as objectId can be same in case of overwrite key

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequestWithFSO.java

@@ -205,7 +205,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
             omMetadataManager, dbOpenKeyToDeleteKey, keyName);
         openKeyToDelete.getMetadata().put(OzoneConsts.OVERWRITTEN_HSYNC_KEY, "true");
         openKeyToDelete.setModificationTime(Time.now());
-        openKeyToDelete.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());
+        openKeyToDelete.setUpdateID(trxnLogIndex);
         OMFileRequest.addOpenFileTableCacheEntry(omMetadataManager,
             dbOpenKeyToDeleteKey, openKeyToDelete, keyName, fileName, trxnLogIndex);
       }
@@ -230,7 +230,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           omKeyInfo.updateLocationInfoList(locationInfoList, false);
 
       // Set the UpdateID to current transactionLogIndex
-      omKeyInfo.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());
+      omKeyInfo.setUpdateID(trxnLogIndex);
 
       // If bucket versioning is turned on during the update, between key
       // creation and key commit, old versions will be just overwritten and
@@ -253,7 +253,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         // Subtract the size of blocks to be overwritten.
         correctedSpace -= keyToDelete.getReplicatedSize();
         RepeatedOmKeyInfo oldVerKeyInfo = getOldVersionsToCleanUp(
-            keyToDelete, trxnLogIndex, ozoneManager.isRatisEnabled());
+            keyToDelete, trxnLogIndex);
         checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
             correctedSpace);
         String delKeyName = omMetadataManager

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java

@@ -277,7 +277,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           keyArgs.getDataSize(), locations, getFileEncryptionInfo(keyArgs),
           ozoneManager.getPrefixManager(), bucketInfo, pathInfo, trxnLogIndex,
           ozoneManager.getObjectIdFromTxId(trxnLogIndex),
-          ozoneManager.isRatisEnabled(), replicationConfig, ozoneManager.getConfiguration());
+          replicationConfig, ozoneManager.getConfiguration());
 
       validateEncryptionKeyInfo(bucketInfo, keyArgs);