HDDS-9233. listStatus() API is not correctly iterating cache.

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

@@ -1426,7 +1426,13 @@ private void listStatusFindKeyInTableCache(
         }
         OzoneFileStatus fileStatus = new OzoneFileStatus(
             cacheOmKeyInfo, scmBlockSize, !OzoneFSUtils.isFile(cacheKey));
-        cacheKeyMap.put(cacheKey, fileStatus);
+        cacheKeyMap.putIfAbsent(cacheKey, fileStatus);
+        // This else block has been added to capture deleted entries in cache.
+        // Adding deleted entries in cacheKeyMap as there is a possible race
+        // condition where table cache iterator is flushed already when
+        // using in the caller of this method.
+      } else if (cacheOmKeyInfo == null && !cacheKeyMap.containsKey(cacheKey)) {
+        cacheKeyMap.put(cacheKey, null);
       }
     }
   }
@@ -1540,8 +1546,14 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
     countEntries = 0;
     // Convert results in cacheKeyMap to List
     for (OzoneFileStatus fileStatus : cacheKeyMap.values()) {
-      // No need to check if a key is deleted or not here, this is handled
-      // when adding entries to cacheKeyMap from DB.
+      // Here need to check if a key is deleted as cacheKeyMap will contain
+      // deleted entries as well. Adding deleted entries in cacheKeyMap is done
+      // as there is a possible race condition where table cache iterator is
+      // flushed already and isKeyDeleted check may not work as expected
+      // before putting entries in cacheKeyMap in findKeyInDbWithIterator call.
+      if (fileStatus == null) {
+        continue;
+      }
       fileStatusList.add(fileStatus);
       countEntries++;
       if (countEntries >= numEntries) {
@@ -1611,9 +1623,8 @@ private void findKeyInDbWithIterator(boolean recursive, String startKey,
           String entryKeyName = omKeyInfo.getKeyName();
           if (recursive) {
             // for recursive list all the entries
-
-            if (!isKeyDeleted(entryInDb, keyTable)) {
-              cacheKeyMap.putIfAbsent(entryInDb, new OzoneFileStatus(omKeyInfo,
+            if (!cacheKeyMap.containsKey(entryInDb)) {
+              cacheKeyMap.put(entryInDb, new OzoneFileStatus(omKeyInfo,
                   scmBlockSize, !OzoneFSUtils.isFile(entryKeyName)));
               countEntries++;
             }
@@ -1626,14 +1637,14 @@ private void findKeyInDbWithIterator(boolean recursive, String startKey,
                 .getImmediateChild(entryKeyName, keyName);
             boolean isFile = OzoneFSUtils.isFile(immediateChild);
             if (isFile) {
-              if (!isKeyDeleted(entryInDb, keyTable)) {
+              if (!cacheKeyMap.containsKey(entryInDb)) {
                 cacheKeyMap.put(entryInDb,
                     new OzoneFileStatus(omKeyInfo, scmBlockSize, !isFile));
                 countEntries++;
               }
             } else {
               // if entry is a directory
-              if (!isKeyDeleted(entryInDb, keyTable)) {
+              if (!cacheKeyMap.containsKey(entryInDb)) {
                 if (!entryKeyName.equals(immediateChild)) {
                   OmKeyInfo fakeDirEntry = createDirectoryKey(
                       omKeyInfo, immediateChild);

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

@@ -215,7 +215,11 @@ public Collection<OzoneFileStatus> listStatusFSO(OmKeyArgs args,
         HeapEntry entry = heapIterator.next();
         OzoneFileStatus status = entry.getStatus(prefixKey,
             scmBlockSize, volumeName, bucketName, replication);
-        map.putIfAbsent(entry.key, status);
+        // Caution: DO NOT use putIfAbsent. putIfAbsent undesirably overwrites
+        // the value with `status` when the existing value in the map is null.
+        if (!map.containsKey(entry.key)) {
+          map.put(entry.key, status);
+        }
       }
     }