HDDS-10077. Add hsync metadata to hsync'ed keys in OpenKeyTable as well

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

@@ -104,7 +104,6 @@
 import com.google.common.collect.Lists;
 import org.apache.commons.lang3.StringUtils;
 import static org.apache.hadoop.ozone.OzoneConsts.DB_TRANSIENT_MARKER;
-import static org.apache.hadoop.ozone.OzoneConsts.HSYNC_CLIENT_ID;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_DB_NAME;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT;
@@ -1207,10 +1206,8 @@ public ListOpenFilesResult listOpenFiles(BucketLayout bucketLayout,
     //  listKeys do. But that complicates the iteration logic by quite a bit.
     //  And if we do that, we need to refactor listKeys as well to dedup.
 
-    final Table<String, OmKeyInfo> okTable, kTable;
+    final Table<String, OmKeyInfo> okTable;
     okTable = getOpenKeyTable(bucketLayout);
-    // keyTable required to check key hsync metadata. TODO: HDDS-10077
-    kTable = getKeyTable(bucketLayout);
 
     // No lock required since table iterator creates a "snapshot"
     try (TableIterator<String, ? extends KeyValue<String, OmKeyInfo>>
@@ -1228,13 +1225,7 @@ public ListOpenFilesResult listOpenFiles(BucketLayout bucketLayout,
           String dbKey = kv.getKey();
           long clientID = OMMetadataManager.getClientIDFromOpenKeyDBKey(dbKey);
           OmKeyInfo omKeyInfo = kv.getValue();
-
-          // Trim client ID to get the keyTable dbKey
-          int lastSlashIdx = dbKey.lastIndexOf(OM_KEY_PREFIX);
-          String ktDbKey = dbKey.substring(0, lastSlashIdx);
-          // Check whether the key has been hsync'ed by checking keyTable
-          checkAndUpdateKeyHsyncStatus(omKeyInfo, ktDbKey, kTable);
-
+          // Note with HDDS-10077, there is no need to check KeyTable for hsync metadata
           openKeySessionList.add(
               new OpenKeySession(clientID, omKeyInfo,
                   omKeyInfo.getLatestVersionLocations().getVersion()));
@@ -1261,23 +1252,6 @@ public ListOpenFilesResult listOpenFiles(BucketLayout bucketLayout,
         openKeySessionList);
   }
 
-  /**
-   * Check and update OmKeyInfo from OpenKeyTable with hsync status in KeyTable.
-   */
-  private void checkAndUpdateKeyHsyncStatus(OmKeyInfo omKeyInfo,
-                                            String dbKey,
-                                            Table<String, OmKeyInfo> kTable)
-      throws IOException {
-    OmKeyInfo ktOmKeyInfo = kTable.get(dbKey);
-    if (ktOmKeyInfo != null) {
-      // The same key in OpenKeyTable also exists in KeyTable, indicating
-      // the key has been hsync'ed
-      String hsyncClientId = ktOmKeyInfo.getMetadata().get(HSYNC_CLIENT_ID);
-      // Append HSYNC_CLIENT_ID to OmKeyInfo to be returned to the client
-      omKeyInfo.getMetadata().put(HSYNC_CLIENT_ID, hsyncClientId);
-    }
-  }
-
   @Override
   public ListKeysResult listKeys(String volumeName, String bucketName,
                                  String startKey, String keyPrefix, int maxKeys)
@@ -1857,7 +1831,6 @@ public ExpiredOpenKeys getExpiredOpenKeys(Duration expireThreshold,
       final long expiredCreationTimestamp =
           expireThreshold.negated().plusMillis(Time.now()).toMillis();
 
-
       int num = 0;
       while (num < count && keyValueTableIterator.hasNext()) {
         KeyValue<String, OmKeyInfo> openKeyValue = keyValueTableIterator.next();
@@ -1875,8 +1848,7 @@ public ExpiredOpenKeys getExpiredOpenKeys(Duration expireThreshold,
           final String clientIdString
               = dbOpenKeyName.substring(lastPrefix + 1);
 
-          final OmKeyInfo info = kt.get(dbKeyName);
-          final boolean isHsync = java.util.Optional.ofNullable(info)
+          final boolean isHsync = java.util.Optional.of(openKeyInfo)
               .map(WithMetadata::getMetadata)
               .map(meta -> meta.get(OzoneConsts.HSYNC_CLIENT_ID))
               .filter(id -> id.equals(clientIdString))
@@ -1886,6 +1858,7 @@ public ExpiredOpenKeys getExpiredOpenKeys(Duration expireThreshold,
             // add non-hsync'ed keys
             expiredKeys.addOpenKey(openKeyInfo, dbOpenKeyName);
           } else {
+            final OmKeyInfo info = kt.get(dbKeyName);
             // add hsync'ed keys
             final KeyArgs.Builder keyArgs = KeyArgs.newBuilder()
                 .setVolumeName(info.getVolumeName())

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

@@ -22,8 +22,6 @@
 
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSecretManager;
-import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
-import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.audit.OMAction;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
@@ -238,7 +236,7 @@ private RecoverLeaseResponse doWork(OzoneManager ozoneManager,
       openKeyInfo.setModificationTime(Time.now());
       // add to cache.
       omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
-          new CacheKey<>(dbOpenFileKey), CacheValue.get(transactionLogIndex, openKeyInfo));
+          dbOpenFileKey, openKeyInfo, transactionLogIndex);
     }
     // override key name with normalizedKeyPath
     keyInfo.setKeyName(keyName);

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

@@ -42,6 +42,7 @@
 import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.WithMetadata;
 import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
+import org.apache.hadoop.ozone.om.request.util.OmKeyHSyncUtil;
 import org.apache.hadoop.ozone.om.request.validation.RequestFeatureValidator;
 import org.apache.hadoop.ozone.om.request.validation.RequestProcessingPhase;
 import org.apache.hadoop.ozone.om.request.validation.ValidationCondition;
@@ -81,8 +82,7 @@
 public class OMKeyCommitRequest extends OMKeyRequest {
 
   @VisibleForTesting
-  public static final Logger LOG =
-      LoggerFactory.getLogger(OMKeyCommitRequest.class);
+  public static final Logger LOG = LoggerFactory.getLogger(OMKeyCommitRequest.class);
 
   public OMKeyCommitRequest(OMRequest omRequest, BucketLayout bucketLayout) {
     super(omRequest, bucketLayout);
@@ -237,7 +237,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
 
       final String clientIdString = String.valueOf(writerClientId);
       if (null != keyToDelete) {
-        isPreviousCommitHsync = java.util.Optional.ofNullable(keyToDelete)
+        isPreviousCommitHsync = java.util.Optional.of(keyToDelete)
             .map(WithMetadata::getMetadata)
             .map(meta -> meta.get(OzoneConsts.HSYNC_CLIENT_ID))
             .filter(id -> id.equals(clientIdString))
@@ -263,8 +263,14 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
       omKeyInfo.getMetadata().putAll(KeyValueUtil.getFromProtobuf(
           commitKeyArgs.getMetadataList()));
 
+      // non-null indicates it is necessary to update the open key
+      OmKeyInfo newOpenKeyInfo = null;
+
       if (isHSync) {
-        omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID, clientIdString);
+        if (!OmKeyHSyncUtil.isHSyncedPreviously(omKeyInfo, clientIdString, dbOpenKey)) {
+          omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID, clientIdString);
+          newOpenKeyInfo = omKeyInfo.copyObject();
+        }
       } else if (isRecovery) {
         omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
         omKeyInfo.getMetadata().remove(OzoneConsts.LEASE_RECOVERY);
@@ -337,6 +343,13 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
         // So that this key can't be committed again.
         omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
             dbOpenKey, trxnLogIndex);
+
+        // Prevent hsync metadata from getting committed to the final key
+        omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
+      } else if (newOpenKeyInfo != null) {
+        // isHSync is true and newOpenKeyInfo is set, update OpenKeyTable
+        omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
+            dbOpenKey, newOpenKeyInfo, trxnLogIndex);
       }
 
       omMetadataManager.getKeyTable(getBucketLayout()).addCacheEntry(
@@ -346,7 +359,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
 
       omClientResponse = new OMKeyCommitResponse(omResponse.build(),
           omKeyInfo, dbOzoneKey, dbOpenKey, omBucketInfo.copyObject(),
-          oldKeyVersionsToDeleteMap, isHSync);
+          oldKeyVersionsToDeleteMap, isHSync, newOpenKeyInfo);
 
       result = Result.SUCCESS;
     } catch (IOException | InvalidPathException ex) {

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

@@ -22,6 +22,7 @@
 import java.util.HashMap;
 
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.ozone.om.request.util.OmKeyHSyncUtil;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.audit.AuditLogger;
@@ -173,8 +174,16 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
 
       omKeyInfo.getMetadata().putAll(KeyValueUtil.getFromProtobuf(
           commitKeyArgs.getMetadataList()));
+
+      final String clientIdString = String.valueOf(writerClientId);
+      // non-null indicates it is necessary to update the open key
+      OmKeyInfo newOpenKeyInfo = null;
+
       if (isHSync) {
-        omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID, String.valueOf(writerClientId));
+        if (!OmKeyHSyncUtil.isHSyncedPreviously(omKeyInfo, clientIdString, dbOpenFileKey)) {
+          omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID, clientIdString);
+          newOpenKeyInfo = omKeyInfo.copyObject();
+        }
       } else if (isRecovery) {
         omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
         omKeyInfo.getMetadata().remove(OzoneConsts.LEASE_RECOVERY);
@@ -196,8 +205,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
       boolean isPreviousCommitHsync = false;
       Map<String, RepeatedOmKeyInfo> oldKeyVersionsToDeleteMap = null;
       if (null != keyToDelete) {
-        final String clientIdString = String.valueOf(writerClientId);
-        isPreviousCommitHsync = java.util.Optional.ofNullable(keyToDelete)
+        isPreviousCommitHsync = java.util.Optional.of(keyToDelete)
             .map(WithMetadata::getMetadata)
             .map(meta -> meta.get(OzoneConsts.HSYNC_CLIENT_ID))
             .filter(id -> id.equals(clientIdString))
@@ -267,6 +275,13 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
         // So that this key can't be committed again.
         OMFileRequest.addOpenFileTableCacheEntry(omMetadataManager,
             dbOpenFileKey, null, fileName, trxnLogIndex);
+
+        // Prevent hsync metadata from getting committed to the final key
+        omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
+      } else if (newOpenKeyInfo != null) {
+        // isHSync is true and newOpenKeyInfo is set, update OpenKeyTable
+        OMFileRequest.addOpenFileTableCacheEntry(omMetadataManager,
+            dbOpenFileKey, newOpenKeyInfo, fileName, trxnLogIndex);
       }
 
       OMFileRequest.addFileTableCacheEntry(omMetadataManager, dbFileKey,
@@ -275,8 +290,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
       omBucketInfo.incrUsedBytes(correctedSpace);
 
       omClientResponse = new OMKeyCommitResponseWithFSO(omResponse.build(),
-              omKeyInfo, dbFileKey, dbOpenFileKey, omBucketInfo.copyObject(),
-          oldKeyVersionsToDeleteMap, volumeId, isHSync);
+          omKeyInfo, dbFileKey, dbOpenFileKey, omBucketInfo.copyObject(),
+          oldKeyVersionsToDeleteMap, volumeId, isHSync, newOpenKeyInfo);
 
       result = Result.SUCCESS;
     } catch (IOException | InvalidPathException ex) {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/util/OmKeyHSyncUtil.java

@@ -0,0 +1,56 @@
+/**
+ * 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.om.request.util;
+
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Helper methods related to OM key HSync.
+ */
+public final class OmKeyHSyncUtil {
+
+  public static final Logger LOG = LoggerFactory.getLogger(OmKeyHSyncUtil.class);
+
+  private OmKeyHSyncUtil() {
+  }
+
+  /**
+   * Returns true if the key has been hsync'ed before (has metadata HSYNC_CLIENT_ID).
+   * @param omKeyInfo OmKeyInfo
+   * @param clientIdString Client ID String
+   * @param dbOpenKey dbOpenKey
+   */
+  public static boolean isHSyncedPreviously(OmKeyInfo omKeyInfo, String clientIdString, String dbOpenKey) {
+    // Check whether the key has been hsync'ed before
+    final String previousHsyncClientId = omKeyInfo.getMetadata().get(OzoneConsts.HSYNC_CLIENT_ID);
+    if (previousHsyncClientId != null) {
+      if (clientIdString.equals(previousHsyncClientId)) {
+        // Same client ID, no need to update OpenKeyTable. One less DB write
+        return true;
+      } else {
+        // Sanity check. Should never enter
+        LOG.warn("Client ID '{}' currently hsync'ing key does not match previous hsync client ID '{}'. dbOpenKey='{}'",
+            clientIdString, previousHsyncClientId, dbOpenKey);
+      }
+    }
+    return false;
+  }
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponse.java

@@ -49,22 +49,25 @@ public class OMKeyCommitResponse extends OmKeyResponse {
   private String openKeyName;
   private OmBucketInfo omBucketInfo;
   private Map<String, RepeatedOmKeyInfo> keyToDeleteMap;
-
   private boolean isHSync;
+  private OmKeyInfo newOpenKeyInfo;
 
+  @SuppressWarnings("checkstyle:ParameterNumber")
   public OMKeyCommitResponse(
       @Nonnull OMResponse omResponse,
       @Nonnull OmKeyInfo omKeyInfo, String ozoneKeyName, String openKeyName,
       @Nonnull OmBucketInfo omBucketInfo,
       Map<String, RepeatedOmKeyInfo> keyToDeleteMap,
-      boolean isHSync) {
+      boolean isHSync,
+      OmKeyInfo newOpenKeyInfo) {
     super(omResponse, omBucketInfo.getBucketLayout());
     this.omKeyInfo = omKeyInfo;
     this.ozoneKeyName = ozoneKeyName;
     this.openKeyName = openKeyName;
     this.omBucketInfo = omBucketInfo;
     this.keyToDeleteMap = keyToDeleteMap;
     this.isHSync = isHSync;
+    this.newOpenKeyInfo = newOpenKeyInfo;
   }
 
   /**
@@ -85,6 +88,9 @@ public void addToDBBatch(OMMetadataManager omMetadataManager,
     if (!isHSync()) {
       omMetadataManager.getOpenKeyTable(getBucketLayout())
           .deleteWithBatch(batchOperation, openKeyName);
+    } else if (newOpenKeyInfo != null) {
+      omMetadataManager.getOpenKeyTable(getBucketLayout()).putWithBatch(
+          batchOperation, openKeyName, newOpenKeyInfo);
     }
 
     omMetadataManager.getKeyTable(getBucketLayout())
@@ -133,4 +139,8 @@ protected void updateDeletedTable(OMMetadataManager omMetadataManager,
   protected boolean isHSync() {
     return isHSync;
   }
+
+  public OmKeyInfo getNewOpenKeyInfo() {
+    return newOpenKeyInfo;
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponseWithFSO.java

@@ -47,16 +47,17 @@ public class OMKeyCommitResponseWithFSO extends OMKeyCommitResponse {
 
   private long volumeId;
 
-  @SuppressWarnings("parameternumber")
+  @SuppressWarnings("checkstyle:ParameterNumber")
   public OMKeyCommitResponseWithFSO(
       @Nonnull OMResponse omResponse,
       @Nonnull OmKeyInfo omKeyInfo,
       String ozoneKeyName, String openKeyName,
       @Nonnull OmBucketInfo omBucketInfo,
       Map<String, RepeatedOmKeyInfo> deleteKeyMap, long volumeId,
-      boolean isHSync) {
+      boolean isHSync,
+      OmKeyInfo newOpenKeyInfo) {
     super(omResponse, omKeyInfo, ozoneKeyName, openKeyName,
-            omBucketInfo, deleteKeyMap, isHSync);
+            omBucketInfo, deleteKeyMap, isHSync, newOpenKeyInfo);
     this.volumeId = volumeId;
   }
 
@@ -78,6 +79,9 @@ public void addToDBBatch(OMMetadataManager omMetadataManager,
     if (!this.isHSync()) {
       omMetadataManager.getOpenKeyTable(getBucketLayout())
           .deleteWithBatch(batchOperation, getOpenKeyName());
+    } else if (getNewOpenKeyInfo() != null) {
+      omMetadataManager.getOpenKeyTable(getBucketLayout()).putWithBatch(
+          batchOperation, getOpenKeyName(), getNewOpenKeyInfo());
     }
 
     OMFileRequest.addToFileTable(omMetadataManager, batchOperation,

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyCommitRequest.java

@@ -45,14 +45,10 @@
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
-    .CommitKeyRequest;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
-    .KeyArgs;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
-    .KeyLocation;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
-    .OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CommitKeyRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocation;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;