Fix review comments.

Author: lokeshj1703

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

@@ -1360,8 +1360,8 @@ public OzoneFileStatus getFileStatus(OmKeyArgs args) throws IOException {
     String bucketName = args.getBucketName();
     String keyName = args.getKeyName();
 
+    metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
     try {
-      metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
       // Check if this is the root of the filesystem.
       if (keyName.length() == 0) {
         validateBucket(volumeName, bucketName);
@@ -1418,8 +1418,8 @@ public void createDirectory(OmKeyArgs args) throws IOException {
     String bucketName = args.getBucketName();
     String keyName = args.getKeyName();
 
+    metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
     try {
-      metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
 
       // Check if this is the root of the filesystem.
       if (keyName.length() == 0) {
@@ -1430,7 +1430,7 @@ public void createDirectory(OmKeyArgs args) throws IOException {
       OzoneFileStatus status =
           verifyNoFilesInPath(volumeName, bucketName, keyPath, false);
       if (status != null && OzoneFSUtils.pathToKey(status.getPath())
-          .equals(keyPath.toString())) {
+          .equals(keyName)) {
         // if directory already exists
         return;
       }
@@ -1490,9 +1490,8 @@ public OpenKeySession createFile(OmKeyArgs args, boolean isOverWrite,
     String keyName = args.getKeyName();
     OpenKeySession keySession;
 
+    metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
     try {
-      metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
-
       OzoneFileStatus fileStatus;
       try {
         fileStatus = getFileStatus(args);
@@ -1539,8 +1538,8 @@ public OmKeyInfo lookupFile(OmKeyArgs args) throws IOException {
     String bucketName = args.getBucketName();
     String keyName = args.getKeyName();
 
+    metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
     try {
-      metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
       OzoneFileStatus fileStatus = getFileStatus(args);
       if (fileStatus.isFile()) {
         return fileStatus.getKeyInfo();
@@ -1573,8 +1572,8 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
     String keyName = args.getKeyName();
 
     List<OzoneFileStatus> fileStatusList = new ArrayList<>();
+    metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
     try {
-      metadataManager.getLock().acquireBucketLock(volumeName, bucketName);
       if (Strings.isNullOrEmpty(startKey)) {
         OzoneFileStatus fileStatus = getFileStatus(args);
         if (fileStatus.isFile()) {

hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java

@@ -498,20 +498,24 @@ public FileStatus[] listStatus(Path f) throws IOException {
     LOG.trace("listStatus() path:{}", f);
     int numEntries = LISTING_PAGE_SIZE;
     LinkedList<OzoneFileStatus> statuses = new LinkedList<>();
-    List<OzoneFileStatus> tmpStatus;
+    List<OzoneFileStatus> tmpStatusList;
     String startKey = "";
 
     do {
-      tmpStatus = adapter.listStatus(pathToKey(f), false, startKey, numEntries);
-      if (!tmpStatus.isEmpty()) {
+      tmpStatusList =
+          adapter.listStatus(pathToKey(f), false, startKey, numEntries);
+      if (!tmpStatusList.isEmpty()) {
         if (startKey.isEmpty()) {
-          statuses.addAll(tmpStatus);
+          statuses.addAll(tmpStatusList);
         } else {
-          statuses.addAll(tmpStatus.subList(1, tmpStatus.size()));
+          statuses.addAll(tmpStatusList.subList(1, tmpStatusList.size()));
         }
         startKey = pathToKey(statuses.getLast().getPath());
       }
-    } while (tmpStatus.size() == numEntries);
+      // listStatus returns entries numEntries in size if available.
+      // Any lesser number of entries indicate that the required entries have
+      // exhausted.
+    } while (tmpStatusList.size() == numEntries);
 
     for (OzoneFileStatus status : statuses) {
       status.makeQualified(uri, status.getPath().makeQualified(uri, workingDir),

hadoop-ozone/ozonefs/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java

@@ -236,12 +236,10 @@ public void testListStatusOnLargeDirectory() throws Exception {
       paths.add(p.getName());
     }
 
-    // ListStatus on root should return dir1 (even though /dir1 key does not
-    // exist) and dir2 only. dir12 is not an immediate child of root and
-    // hence should not be listed.
     FileStatus[] fileStatuses = o3fs.listStatus(root);
-    assertEquals("FileStatus should return only the immediate children", numDirs,
-        fileStatuses.length);
+    assertEquals(
+        "Total directories listed do not match the existing directories",
+        numDirs, fileStatuses.length);
 
     for (int i=0; i < numDirs; i++) {
       assertTrue(paths.contains(fileStatuses[i].getPath().getName()));