From 5cdd880078f13d08683925c96448299a78aec008 Mon Sep 17 00:00:00 2001
From: Anu Engineer <aengineer@apache.org>
Date: Fri, 2 Jun 2017 00:48:39 -0700
Subject: [PATCH] HDFS-11913. Ozone: TestKeySpaceManager#testDeleteVolume
 fails. Contributed by Weiwei Yang.

---
 .../hadoop/ozone/ksm/MetadataManagerImpl.java | 12 +++--
 .../hadoop/ozone/ksm/TestKeySpaceManager.java | 45 ++++++++++++-------
 2 files changed, 38 insertions(+), 19 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManagerImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManagerImpl.java
index 8efe7240c72..043175efbd0 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManagerImpl.java
+++ b/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;
+      }
     }
   }
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java
index 90dbc253e96..0acebcb59cb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java
+++ b/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,