HDDS-2779. Fix list volume for --start parameter

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/ObjectStore.java

@@ -299,11 +299,6 @@ private class VolumeIterator implements Iterator<OzoneVolume> {
 
     @Override
     public boolean hasNext() {
-      if(!currentIterator.hasNext()) {
-        currentIterator = getNextListOfVolumes(
-            currentValue != null ? currentValue.getName() : null)
-            .iterator();
-      }
       return currentIterator.hasNext();
     }
 

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

@@ -815,34 +815,55 @@ public List<OmVolumeArgs> listVolumes(String userName,
     }
     volumes = getVolumesByUser(userName);
 
-    if (volumes == null || volumes.getVolumeNamesCount() == 0) {
+    if (volumes.getVolumeNamesCount() == 0) {
       return result;
     }
 
+    /* Process listing volumes with startVolume. */
     boolean startKeyFound = Strings.isNullOrEmpty(startKey);
+    if (!startKeyFound) {
+      OmVolumeArgs startVolArgs = getVolumeTable()
+          .get(this.getVolumeKey(startKey));
+
+      if (startVolArgs == null) {
+        throw new OMException("StartVolume info not found for " + startKey,
+            ResultCodes.VOLUME_NOT_FOUND);
+      } else if (!startKey.startsWith(prefix)) {
+        throw new OMException("StartVolume " + startKey + " not found" +
+            " for prefix " + prefix, ResultCodes.VOLUME_NOT_FOUND);
+      }
+
+      result.add(startVolArgs);
+    }
+
+    OmVolumeArgs volumeArgs;
     for (String volumeName : volumes.getVolumeNamesList()) {
+
       if (!Strings.isNullOrEmpty(prefix)) {
         if (!volumeName.startsWith(prefix)) {
           continue;
         }
       }
 
+      volumeArgs = getVolumeTable().get(this.getVolumeKey(volumeName));
+      if (volumeArgs == null) {
+        // Could not get volume info by given volume name,
+        // since the volume name is loaded from db,
+        // this probably means om db is corrupted or some entries are
+        // accidentally removed.
+        throw new OMException("Volume info not found for " + volumeName,
+            ResultCodes.VOLUME_NOT_FOUND);
+      }
+
       if (!startKeyFound && volumeName.equals(startKey)) {
         startKeyFound = true;
         continue;
       }
-      if (startKeyFound && result.size() < maxKeys) {
-        OmVolumeArgs volumeArgs =
-            getVolumeTable().get(this.getVolumeKey(volumeName));
-        if (volumeArgs == null) {
-          // Could not get volume info by given volume name,
-          // since the volume name is loaded from db,
-          // this probably means om db is corrupted or some entries are
-          // accidentally removed.
-          throw new OMException("Volume info not found for " + volumeName,
-              ResultCodes.VOLUME_NOT_FOUND);
-        }
+
+      if (result.size() < maxKeys) {
         result.add(volumeArgs);
+      } else {
+        break;
       }
     }
 

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
 import org.apache.hadoop.ozone.om.request.TestOMRequestUtils;
 import org.junit.Assert;
 import org.junit.Before;
@@ -55,6 +56,36 @@ public void setup() throws Exception {
         folder.getRoot().getAbsolutePath());
     omMetadataManager = new OmMetadataManagerImpl(ozoneConfiguration);
   }
+
+  @Test
+  public void testListVolumes() throws Exception {
+    String ownerName = "owner";
+    OmVolumeArgs.Builder argsBuilder = OmVolumeArgs.newBuilder()
+        .setAdminName("admin")
+        .setOwnerName(ownerName);
+
+    String volName;
+    char postfix = 'a';
+    OmVolumeArgs omVolumeArgs;
+    for (int i = 0; i < 26; i++) {
+      volName = "vol" + (char)(postfix + i);
+      omVolumeArgs = argsBuilder
+          .setVolume(volName)
+          .build();
+
+      TestOMRequestUtils.addVolumeToOM(omMetadataManager, omVolumeArgs);
+      TestOMRequestUtils.addUserToDB(volName, ownerName, omMetadataManager);
+    }
+
+    // Test list volumes with setting startVolume.
+    String prefix = "";
+    String startVolume = "volz";
+    List<OmVolumeArgs> volumeList = omMetadataManager.listVolumes(ownerName,
+        prefix, startVolume, 100);
+    Assert.assertEquals(volumeList.get(0).getVolume(), startVolume);
+    Assert.assertEquals(volumeList.size(), 26);
+  }
+
   @Test
   public void testListBuckets() throws Exception {
 

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/TestOMRequestUtils.java

@@ -302,13 +302,22 @@ public static List< HddsProtos.KeyValue> getMetadataList() {
    */
   public static void addUserToDB(String volumeName, String ownerName,
       OMMetadataManager omMetadataManager) throws Exception {
-    OzoneManagerProtocolProtos.UserVolumeInfo userVolumeInfo =
-        OzoneManagerProtocolProtos.UserVolumeInfo
-            .newBuilder()
-            .addVolumeNames(volumeName)
-            .setObjectID(1)
-            .setUpdateID(1)
-            .build();
+
+    OzoneManagerProtocolProtos.UserVolumeInfo userVolumeInfo = omMetadataManager
+        .getUserTable().get(omMetadataManager.getUserKey(ownerName));
+    if (userVolumeInfo == null) {
+      userVolumeInfo = OzoneManagerProtocolProtos.UserVolumeInfo
+          .newBuilder()
+          .addVolumeNames(volumeName)
+          .setObjectID(1)
+          .setUpdateID(1)
+          .build();
+    } else {
+      userVolumeInfo = userVolumeInfo.toBuilder()
+          .addVolumeNames(volumeName)
+          .build();
+    }
+
     omMetadataManager.getUserTable().put(
         omMetadataManager.getUserKey(ownerName), userVolumeInfo);
   }