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,11 @@ private void listStatusFindKeyInTableCache(
         }
         OzoneFileStatus fileStatus = new OzoneFileStatus(
             cacheOmKeyInfo, scmBlockSize, !OzoneFSUtils.isFile(cacheKey));
-        cacheKeyMap.put(cacheKey, fileStatus);
+        cacheKeyMap.putIfAbsent(cacheKey, fileStatus);
+      } else if (cacheOmKeyInfo == null
+          && cacheKey.startsWith(startCacheKey)
+          && cacheKey.compareTo(startCacheKey) >= 0) {
+        cacheKeyMap.putIfAbsent(cacheKey, null);
       }
     }
   }
@@ -1540,8 +1544,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,8 +1621,9 @@ private void findKeyInDbWithIterator(boolean recursive, String startKey,
           String entryKeyName = omKeyInfo.getKeyName();
           if (recursive) {
             // for recursive list all the entries
-
-            if (!isKeyDeleted(entryInDb, keyTable)) {
+            // Since putIfAbsent doesn't work as expected in case of null value,
+            // so had to explicitly check using containsKey
+            if (!cacheKeyMap.containsKey(entryInDb)) {
               cacheKeyMap.putIfAbsent(entryInDb, new OzoneFileStatus(omKeyInfo,
                   scmBlockSize, !OzoneFSUtils.isFile(entryKeyName)));
               countEntries++;
@@ -1626,23 +1637,27 @@ private void findKeyInDbWithIterator(boolean recursive, String startKey,
                 .getImmediateChild(entryKeyName, keyName);
             boolean isFile = OzoneFSUtils.isFile(immediateChild);
             if (isFile) {
-              if (!isKeyDeleted(entryInDb, keyTable)) {
-                cacheKeyMap.put(entryInDb,
+              // Since putIfAbsent doesn't work as expected in case of null
+              // value, so had to explicitly check using containsKey
+              if (!cacheKeyMap.containsKey(entryInDb)) {
+                cacheKeyMap.putIfAbsent(entryInDb,
                     new OzoneFileStatus(omKeyInfo, scmBlockSize, !isFile));
                 countEntries++;
               }
             } else {
               // if entry is a directory
-              if (!isKeyDeleted(entryInDb, keyTable)) {
+              // Since putIfAbsent doesn't work as expected in case of null
+              // value, so had to explicitly check using containsKey
+              if (!cacheKeyMap.containsKey(entryInDb)) {
                 if (!entryKeyName.equals(immediateChild)) {
                   OmKeyInfo fakeDirEntry = createDirectoryKey(
                       omKeyInfo, immediateChild);
-                  cacheKeyMap.put(entryInDb,
+                  cacheKeyMap.putIfAbsent(entryInDb,
                       new OzoneFileStatus(fakeDirEntry,
                           scmBlockSize, true));
                 } else {
                   // If entryKeyName matches dir name, we have the info
-                  cacheKeyMap.put(entryInDb,
+                  cacheKeyMap.putIfAbsent(entryInDb,
                       new OzoneFileStatus(omKeyInfo, 0, true));
                 }
                 countEntries++;

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);
+        // Since putIfAbsent doesn't work as expected in case of null value,
+        // so had to explicitly check using containsKey
+        if (!map.containsKey(entry.key)) {
+          map.putIfAbsent(entry.key, status);
+        }
       }
     }