HDDS-1638. Implement Key Write Requests to use Cache and DoubleBuffer. (apache#956)

Author: bharatviswa504

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/Table.java

@@ -123,6 +123,14 @@ default void addCacheEntry(CacheKey<KEY> cacheKey,
     throw new NotImplementedException("addCacheEntry is not implemented");
   }
 
+  /**
+   * Get the cache value from table cache.
+   * @param cacheKey
+   */
+  default CacheValue<VALUE> getCacheValue(CacheKey<KEY> cacheKey) {
+    throw new NotImplementedException("getCacheValue is not implemented");
+  }
+
   /**
    * Removes all the entries from the table cache which are having epoch value
    * less

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/TypedTable.java

@@ -158,6 +158,11 @@ public void addCacheEntry(CacheKey<KEY> cacheKey,
     cache.put(cacheKey, cacheValue);
   }
 
+  @Override
+  public CacheValue<VALUE> getCacheValue(CacheKey<KEY> cacheKey) {
+    return cache.get(cacheKey);
+  }
+
   public Iterator<Map.Entry<CacheKey<KEY>, CacheValue<VALUE>>> cacheIterator() {
     return cache.iterator();
   }

hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/audit/DummyAction.java

@@ -24,7 +24,6 @@ public enum DummyAction implements AuditAction {
 
   CREATE_VOLUME,
   CREATE_BUCKET,
-  CREATE_KEY,
   READ_VOLUME,
   READ_BUCKET,
   READ_KEY,

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/audit/OMAction.java

@@ -30,7 +30,6 @@ public enum OMAction implements AuditAction {
   COMMIT_KEY,
   CREATE_VOLUME,
   CREATE_BUCKET,
-  CREATE_KEY,
   DELETE_VOLUME,
   DELETE_BUCKET,
   DELETE_KEY,

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

@@ -165,28 +165,34 @@ public void updateLocationInfoList(List<OmKeyLocationInfo> locationInfoList) {
    * part of the latest version, not a new version.
    *
    * @param newLocationList the list of new blocks to be added.
+   * @param updateTime if true, will update modification time.
    * @throws IOException
    */
   public synchronized void appendNewBlocks(
-      List<OmKeyLocationInfo> newLocationList) throws IOException {
+      List<OmKeyLocationInfo> newLocationList, boolean updateTime)
+      throws IOException {
     if (keyLocationVersions.size() == 0) {
       throw new IOException("Appending new block, but no version exist");
     }
     OmKeyLocationInfoGroup currentLatestVersion =
         keyLocationVersions.get(keyLocationVersions.size() - 1);
     currentLatestVersion.appendNewBlocks(newLocationList);
-    setModificationTime(Time.now());
+    if (updateTime) {
+      setModificationTime(Time.now());
+    }
   }
 
   /**
    * Add a new set of blocks. The new blocks will be added as appending a new
    * version to the all version list.
    *
    * @param newLocationList the list of new blocks to be added.
+   * @param updateTime - if true, updates modification time.
    * @throws IOException
    */
   public synchronized long addNewVersion(
-      List<OmKeyLocationInfo> newLocationList) throws IOException {
+      List<OmKeyLocationInfo> newLocationList, boolean updateTime)
+      throws IOException {
     long latestVersionNum;
     if (keyLocationVersions.size() == 0) {
       // no version exist, these blocks are the very first version.
@@ -202,7 +208,10 @@ public synchronized long addNewVersion(
       keyLocationVersions.add(newVersion);
       latestVersionNum = newVersion.getVersion();
     }
-    setModificationTime(Time.now());
+
+    if (updateTime) {
+      setModificationTime(Time.now());
+    }
     return latestVersionNum;
   }
 

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

@@ -23,6 +23,8 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocation;
 import org.apache.hadoop.security.token.Token;
 
+import java.util.Objects;
+
 /**
  * One key can be too huge to fit in one container. In which case it gets split
  * into a number of subkeys. This class represents one such subkey instance.
@@ -202,4 +204,27 @@ public String toString() {
         ", pipeline=" + pipeline +
         ", createVersion=" + createVersion + '}';
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    OmKeyLocationInfo that = (OmKeyLocationInfo) o;
+    return length == that.length &&
+        offset == that.offset &&
+        createVersion == that.createVersion &&
+        Objects.equals(blockID, that.blockID) &&
+        Objects.equals(token, that.token) &&
+        Objects.equals(pipeline, that.pipeline);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(blockID, length, offset, token, createVersion,
+        pipeline);
+  }
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -726,11 +726,20 @@ private OMResponse handleError(OMResponse resp) throws OMException {
   public OmKeyLocationInfo allocateBlock(OmKeyArgs args, long clientId,
       ExcludeList excludeList) throws IOException {
     AllocateBlockRequest.Builder req = AllocateBlockRequest.newBuilder();
-    KeyArgs keyArgs = KeyArgs.newBuilder()
+    KeyArgs.Builder keyArgs = KeyArgs.newBuilder()
         .setVolumeName(args.getVolumeName())
         .setBucketName(args.getBucketName())
         .setKeyName(args.getKeyName())
-        .setDataSize(args.getDataSize()).build();
+        .setDataSize(args.getDataSize());
+
+    if (args.getFactor() != null) {
+      keyArgs.setFactor(args.getFactor());
+    }
+
+    if (args.getType() != null) {
+      keyArgs.setType(args.getType());
+    }
+
     req.setKeyArgs(keyArgs);
     req.setClientID(clientId);
     req.setExcludeList(excludeList.getProtoBuf());

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

@@ -617,6 +617,10 @@ message KeyArgs {
     optional uint32 multipartNumber = 10;
     repeated hadoop.hdds.KeyValue metadata = 11;
     repeated OzoneAclInfo acls = 12;
+    // This will be set when the request is received in pre-Execute. This
+    // value is used in setting creation/modification time depending on the
+    // request type.
+    optional uint64 modificationTime = 13;
 }
 
 message KeyLocation {
@@ -712,6 +716,9 @@ message ListStatusResponse {
 
 message CreateKeyRequest {
     required KeyArgs keyArgs = 1;
+    // Set in OM HA during preExecute step. This way all OM's use same ID in
+    // OM HA.
+    optional uint64 clientID = 2;
 }
 
 message CreateKeyResponse {

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java

@@ -103,7 +103,7 @@ public void restartOzoneManager() throws IOException {
   public void stop() {
     for (OzoneManager ozoneManager : ozoneManagers) {
       if (ozoneManager != null) {
-        LOG.info("Stopping the OzoneManager " + ozoneManager.getOMNodId());
+        LOG.info("Stopping the OzoneManager " + ozoneManager.getOMNodeId());
         ozoneManager.stop();
         ozoneManager.join();
       }

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

@@ -479,7 +479,7 @@ public void testOMProxyProviderInitialization() throws Exception {
         }
       }
       Assert.assertTrue("There is no OM Client Proxy corresponding to OM " +
-              "node" + cluster.getOzoneManager(i).getOMNodId(),
+              "node" + cluster.getOzoneManager(i).getOMNodeId(),
           omClientProxyExists);
     }
   }
@@ -604,7 +604,7 @@ public void testReadRequest() throws Exception {
 
       // Failover to the OM node that the objectStore points to
       omFailoverProxyProvider.performFailoverIfRequired(
-          ozoneManager.getOMNodId());
+          ozoneManager.getOMNodeId());
 
       // A read request should result in the proxyProvider failing over to
       // leader node.