HDFS-11913. Ozone: TestKeySpaceManager#testDeleteVolume fails. Contributed by Weiwei Yang.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManagerImpl.java

@@ -201,10 +201,14 @@ public class MetadataManagerImpl implements  MetadataManager {
       byte[] dbVolumeRootKey = DFSUtil.string2Bytes(dbVolumeRootName);
       // Seek to the root of the volume and look for the next key
       iterator.seek(dbVolumeRootKey);
-      String firstBucketKey = DFSUtil.bytes2String(iterator.next().getKey());
-      // if the key starts with /<volume name>
-      // then there is at least one bucket
-      return !firstBucketKey.startsWith(OzoneConsts.KSM_VOLUME_PREFIX + volume);
+      if (iterator.hasNext()) {
+        String firstBucketKey = DFSUtil.bytes2String(iterator.next().getKey());
+        // if the key starts with /<volume name>/
+        // then there is at least one bucket
+        return !firstBucketKey.startsWith(dbVolumeRootName);
+      } else {
+        return true;
+      }
     }
   }
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java

@@ -187,26 +187,41 @@ public class TestKeySpaceManager {
     String userName = "user" + RandomStringUtils.randomNumeric(5);
     String adminName = "admin" + RandomStringUtils.randomNumeric(5);
     String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
+    String volumeName1 = volumeName + "_A";
+    String volumeName2 = volumeName + "_AA";
+    VolumeArgs volumeArgs = null;
+    VolumeInfo volumeInfo = null;
 
-    VolumeArgs createVolumeArgs = new VolumeArgs(volumeName, userArgs);
-    createVolumeArgs.setUserName(userName);
-    createVolumeArgs.setAdminName(adminName);
-    storageHandler.createVolume(createVolumeArgs);
+    // Create 2 empty volumes with same prefix.
+    volumeArgs = new VolumeArgs(volumeName1, userArgs);
+    volumeArgs.setUserName(userName);
+    volumeArgs.setAdminName(adminName);
+    storageHandler.createVolume(volumeArgs);
 
-    VolumeArgs getVolumeArgs = new VolumeArgs(volumeName, userArgs);
-    VolumeInfo retVolumeInfo = storageHandler.getVolumeInfo(getVolumeArgs);
-    Assert.assertTrue(retVolumeInfo.getVolumeName().equals(volumeName));
-    Assert.assertTrue(retVolumeInfo.getOwner().getName().equals(userName));
+    volumeArgs = new VolumeArgs(volumeName2, userArgs);
+    volumeArgs.setUserName(userName);
+    volumeArgs.setAdminName(adminName);
+    storageHandler.createVolume(volumeArgs);
+
+    volumeArgs  = new VolumeArgs(volumeName1, userArgs);
+    volumeInfo = storageHandler.getVolumeInfo(volumeArgs);
+    Assert.assertTrue(volumeInfo.getVolumeName().equals(volumeName1));
+    Assert.assertTrue(volumeInfo.getOwner().getName().equals(userName));
     Assert.assertEquals(0, ksmMetrics.getNumVolumeCreateFails());
 
-    storageHandler.deleteVolume(createVolumeArgs);
+    // Volume with _A should be able to delete as it is empty.
+    storageHandler.deleteVolume(volumeArgs);
 
-    try {
-      retVolumeInfo = storageHandler.getVolumeInfo(getVolumeArgs);
-    } catch (IOException ex) {
-      Assert.assertEquals(ex.getMessage(),
-          "Info Volume failed, error:VOLUME_NOT_FOUND");
-    }
+    // Make sure volume with _AA suffix still exists.
+    volumeArgs = new VolumeArgs(volumeName2, userArgs);
+    volumeInfo = storageHandler.getVolumeInfo(volumeArgs);
+    Assert.assertTrue(volumeInfo.getVolumeName().equals(volumeName2));
+
+    // Make sure volume with _A suffix is successfully deleted.
+    exception.expect(IOException.class);
+    exception.expectMessage("Info Volume failed, error:VOLUME_NOT_FOUND");
+    volumeArgs = new VolumeArgs(volumeName1, userArgs);
+    storageHandler.getVolumeInfo(volumeArgs);
   }
 
   // Create a volume and a bucket inside the volume,