HDDS-10650. Delete hsync key info from openFileTable while deleting directory recursively.

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestHSync.java

@@ -25,7 +25,9 @@
 import java.nio.charset.StandardCharsets;
 import java.security.GeneralSecurityException;
 import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
@@ -113,6 +115,8 @@
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_DEFAULT_BUCKET_LAYOUT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_ENABLE_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -155,6 +159,7 @@ public static void init() throws Exception {
     CONF.setInt(OZONE_SCM_RATIS_PIPELINE_LIMIT, 10);
     // Reduce KeyDeletingService interval
     CONF.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100, TimeUnit.MILLISECONDS);
+    CONF.setTimeDuration(OZONE_DIR_DELETING_SERVICE_INTERVAL, 100, TimeUnit.MILLISECONDS);
     CONF.setBoolean("ozone.client.incremental.chunk.list", true);
     CONF.setBoolean(OZONE_CHUNK_LIST_INCREMENTAL, true);
     ClientConfigForTesting.newBuilder(StorageUnit.BYTES)
@@ -392,6 +397,54 @@ public void testHSyncDeletedKey() throws Exception {
     }
   }
 
+  @Test
+  public void testHSyncOpenKeyDeletionWhileDeleteDirectory() throws Exception {
+    // Verify that when directory is deleted recursively hsync related openKeys should be deleted,
+
+    // Set the fs.defaultFS
+    final String rootPath = String.format("%s://%s/",
+        OZONE_OFS_URI_SCHEME, CONF.get(OZONE_OM_ADDRESS_KEY));
+    CONF.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, rootPath);
+
+    final String dir = OZONE_ROOT + bucket.getVolumeName()
+        + OZONE_URI_DELIMITER + bucket.getName() + OZONE_URI_DELIMITER + "dir1/dir2";
+    final Path key1 = new Path(dir, "hsync-key");
+
+    try (FileSystem fs = FileSystem.get(CONF)) {
+      // Create key1
+      try (FSDataOutputStream os = fs.create(key1, true)) {
+        os.write(1);
+        os.hsync();
+        // There should be 1 key in openFileTable
+        assertThat(1 == getOpenKeyInfo(BucketLayout.FILE_SYSTEM_OPTIMIZED).size());
+        // Delete directory recursively
+        fs.delete(new Path(OZONE_ROOT + bucket.getVolumeName() + OZONE_URI_DELIMITER +
+            bucket.getName() + OZONE_URI_DELIMITER + "dir1/"), true);
+
+        // Verify entry from openKey gets deleted eventually
+        GenericTestUtils.waitFor(() ->
+            0 == getOpenKeyInfo(BucketLayout.FILE_SYSTEM_OPTIMIZED).size(), 1000, 12000);
+      } catch (OMException ex) {
+        assertEquals(OMException.ResultCodes.DIRECTORY_NOT_FOUND, ex.getResult());
+      }
+    }
+  }
+
+  private List<OmKeyInfo> getOpenKeyInfo(BucketLayout bucketLayout) {
+    List<OmKeyInfo> omKeyInfo = new ArrayList<>();
+
+    Table<String, OmKeyInfo> openFileTable =
+        cluster.getOzoneManager().getMetadataManager().getOpenKeyTable(bucketLayout);
+    try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>>
+             iterator = openFileTable.iterator()) {
+      while (iterator.hasNext()) {
+        omKeyInfo.add(iterator.next().getValue());
+      }
+    } catch (Exception e) {
+    }
+    return omKeyInfo;
+  }
+
   @Test
   public void testUncommittedBlocks() throws Exception {
     // Set the fs.defaultFS

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

@@ -21,9 +21,12 @@
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.om.OMMetrics;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
@@ -39,8 +42,6 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
 
-import java.util.List;
-
 import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.BUCKET_LOCK;
 
 /**
@@ -66,6 +67,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
     Set<Pair<String, String>> lockSet = new HashSet<>();
     Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap = new HashMap<>();
     OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager();
+    List<String> openKeyList = new ArrayList<>();
 
     OMMetrics omMetrics = ozoneManager.getMetrics();
     try {
@@ -110,6 +112,17 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
                 volumeName, bucketName);
             lockSet.add(volBucketPair);
           }
+
+          // If omKeyInfo has hsync metadata, delete its corresponding open key as well
+          String dbOpenKey;
+          String hsyncClientId = keyInfo.getMetadata().get(OzoneConsts.HSYNC_CLIENT_ID);
+          if (hsyncClientId != null) {
+            long parentId = keyInfo.getParentObjectID();
+            dbOpenKey = omMetadataManager.getOpenFileName(path.getVolumeId(), path.getBucketId(),
+                parentId, keyInfo.getFileName(), hsyncClientId);
+            openKeyList.add(dbOpenKey);
+          }
+
           omMetrics.decNumKeys();
           OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
               volumeName, bucketName);
@@ -142,7 +155,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
         getOmRequest());
     OMClientResponse omClientResponse = new OMDirectoriesPurgeResponseWithFSO(
         omResponse.build(), purgeRequests, ozoneManager.isRatisEnabled(),
-            getBucketLayout(), volBucketInfoMap, fromSnapshotInfo);
+            getBucketLayout(), volBucketInfoMap, fromSnapshotInfo, openKeyList);
 
     return omClientResponse;
   }

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

@@ -62,17 +62,19 @@ public class OMDirectoriesPurgeResponseWithFSO extends OmKeyResponse {
   private boolean isRatisEnabled;
   private Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap;
   private SnapshotInfo fromSnapshotInfo;
+  private List<String> openKeyList;
 
   public OMDirectoriesPurgeResponseWithFSO(@Nonnull OMResponse omResponse,
       @Nonnull List<OzoneManagerProtocolProtos.PurgePathRequest> paths,
       boolean isRatisEnabled, @Nonnull BucketLayout bucketLayout,
       Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap,
-      SnapshotInfo fromSnapshotInfo) {
+      SnapshotInfo fromSnapshotInfo, List<String> openKeyList) {
     super(omResponse, bucketLayout);
     this.paths = paths;
     this.isRatisEnabled = isRatisEnabled;
     this.volBucketInfoMap = volBucketInfoMap;
     this.fromSnapshotInfo = fromSnapshotInfo;
+    this.openKeyList = openKeyList;
   }
 
   @Override
@@ -165,6 +167,13 @@ public void processPaths(OMMetadataManager omMetadataManager,
             deletedKey, repeatedOmKeyInfo);
       }
 
+      if (!openKeyList.isEmpty()) {
+        for (String openKey : openKeyList) {
+          omMetadataManager.getOpenKeyTable(getBucketLayout()).deleteWithBatch(
+              batchOperation, openKey);
+        }
+      }
+
       // Delete the visited directory from deleted directory table
       if (path.hasDeletedDir()) {
         omMetadataManager.getDeletedDirTable().deleteWithBatch(batchOperation,