getBatchedListing throws ObserverRetryOnActiveException eagerly

Author: dParikesit

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -4325,11 +4325,22 @@ BatchedDirectoryListing getBatchedListing(String[] srcs, byte[] startAfter,
           if (dirListing == null) {
             throw new FileNotFoundException("Path " + src + " does not exist");
           }
+          if (needLocation && isObserver()) {
+            for (HdfsFileStatus fs : dirListing.getPartialListing()) {
+              if (fs instanceof HdfsLocatedFileStatus) {
+                LocatedBlocks lbs = ((HdfsLocatedFileStatus) fs).getLocatedBlocks();
+                checkBlockLocationsWhenObserver(lbs, fs.toString());
+              }
+            }
+          }
           listing = new HdfsPartialListing(
               srcsIndex, Lists.newArrayList(dirListing.getPartialListing()));
           numEntries += listing.getPartialListing().size();
           lastListing = dirListing;
         } catch (Exception e) {
+          if (e instanceof ObserverRetryOnActiveException) {
+            throw (ObserverRetryOnActiveException) e;
+          }
           if (e instanceof AccessControlException) {
             logAuditEvent(false, operationName, src);
           }
@@ -4392,6 +4403,8 @@ BatchedDirectoryListing getBatchedListing(String[] srcs, byte[] startAfter,
             true,
             returnedStartAfter);
       }
+    } catch(ObserverRetryOnActiveException e){
+      throw e;
     } finally {
       readUnlock(RwLockMode.FS, operationName,
           getLockReportInfoSupplier(Arrays.toString(srcs)));