HDDS-1685. Recon: Add support for "start" query param to containers and containers/{id} endpoints.

hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconConstants.java

@@ -38,4 +38,8 @@ public final class ReconConstants {
   public static final String CONTAINER_KEY_TABLE =
       "containerKeyTable";
 
+  public static final String FETCH_ALL = "-1";
+  public static final String RECON_QUERY_PREVKEY = "prev-key";
+  public static final String RECON_QUERY_LIMIT = "limit";
+
 }

hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/api/ContainerKeyService.java

@@ -38,6 +38,7 @@ import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
@@ -50,6 +51,10 @@ import org.apache.hadoop.ozone.recon.spi.ContainerDBServiceProvider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.ozone.recon.ReconConstants.FETCH_ALL;
+import static org.apache.hadoop.ozone.recon.ReconConstants.RECON_QUERY_LIMIT;
+import static org.apache.hadoop.ozone.recon.ReconConstants.RECON_QUERY_PREVKEY;
+
 
 /**
  * Endpoint for querying keys that belong to a container.
@@ -69,16 +74,20 @@ public class ContainerKeyService {
 
   /**
    * Return @{@link org.apache.hadoop.ozone.recon.api.types.ContainerMetadata}
-   * for all the containers.
+   * for the containers starting from the given "prev-key" query param for the
+   * given "limit". The given "prev-key" is skipped from the results returned.
    *
+   * @param limit max no. of containers to get.
+   * @param prevKey the containerID after which results are returned.
    * @return {@link Response}
    */
   @GET
   public Response getContainers(
-      @DefaultValue("-1") @QueryParam("limit") int limit) {
+      @DefaultValue(FETCH_ALL) @QueryParam(RECON_QUERY_LIMIT) int limit,
+      @DefaultValue("0") @QueryParam(RECON_QUERY_PREVKEY) long prevKey) {
     Map<Long, ContainerMetadata> containersMap;
     try {
-      containersMap = containerDBServiceProvider.getContainers(limit);
+      containersMap = containerDBServiceProvider.getContainers(limit, prevKey);
     } catch (IOException ioEx) {
       throw new WebApplicationException(ioEx,
           Response.Status.INTERNAL_SERVER_ERROR);
@@ -88,20 +97,27 @@ public class ContainerKeyService {
 
   /**
    * Return @{@link org.apache.hadoop.ozone.recon.api.types.KeyMetadata} for
-   * all keys that belong to the container identified by the id param.
+   * all keys that belong to the container identified by the id param
+   * starting from the given "prev-key" query param for the given "limit".
+   * The given prevKeyPrefix is skipped from the results returned.
    *
-   * @param containerId Container Id
+   * @param containerID the given containerID.
+   * @param limit max no. of keys to get.
+   * @param prevKeyPrefix the key prefix after which results are returned.
    * @return {@link Response}
    */
   @GET
   @Path("/{id}")
   public Response getKeysForContainer(
-      @PathParam("id") Long containerId,
-      @DefaultValue("-1") @QueryParam("limit") int limit) {
+      @PathParam("id") Long containerID,
+      @DefaultValue(FETCH_ALL) @QueryParam(RECON_QUERY_LIMIT) int limit,
+      @DefaultValue(StringUtils.EMPTY) @QueryParam(RECON_QUERY_PREVKEY)
+          String prevKeyPrefix) {
     Map<String, KeyMetadata> keyMetadataMap = new LinkedHashMap<>();
     try {
       Map<ContainerKeyPrefix, Integer> containerKeyPrefixMap =
-          containerDBServiceProvider.getKeyPrefixesForContainer(containerId);
+          containerDBServiceProvider.getKeyPrefixesForContainer(containerID,
+              prevKeyPrefix);
 
       // Get set of Container-Key mappings for given containerId.
       for (ContainerKeyPrefix containerKeyPrefix : containerKeyPrefixMap
@@ -128,7 +144,7 @@ public class ContainerKeyService {
           List<OmKeyLocationInfo> omKeyLocationInfos = omKeyLocationInfoGroup
               .getLocationList()
               .stream()
-              .filter(c -> c.getContainerID() == containerId)
+              .filter(c -> c.getContainerID() == containerID)
               .collect(Collectors.toList());
           for (OmKeyLocationInfo omKeyLocationInfo : omKeyLocationInfos) {
             blockIds.add(new ContainerBlockMetadata(omKeyLocationInfo

hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/ContainerDBServiceProvider.java

@@ -66,26 +66,32 @@ public interface ContainerDBServiceProvider {
    * @param containerId the given containerId.
    * @return Map of Key prefix -> count.
    */
-  Map<ContainerKeyPrefix, Integer> getKeyPrefixesForContainer(long containerId)
-      throws IOException;
+  Map<ContainerKeyPrefix, Integer> getKeyPrefixesForContainer(
+      long containerId) throws IOException;
 
   /**
-   * Get a Map of containerID, containerMetadata of all the Containers.
+   * Get the stored key prefixes for the given containerId starting
+   * after the given keyPrefix.
    *
-   * @return Map of containerID -> containerMetadata.
-   * @throws IOException
+   * @param containerId the given containerId.
+   * @param prevKeyPrefix the key prefix to seek to and start scanning.
+   * @return Map of Key prefix -> count.
    */
-  Map<Long, ContainerMetadata> getContainers() throws IOException;
+  Map<ContainerKeyPrefix, Integer> getKeyPrefixesForContainer(
+      long containerId, String prevKeyPrefix) throws IOException;
 
   /**
    * Get a Map of containerID, containerMetadata of Containers only for the
    * given limit. If the limit is -1 or any integer <0, then return all
    * the containers without any limit.
    *
+   * @param limit the no. of containers to fetch.
+   * @param prevContainer containerID after which the results are returned.
    * @return Map of containerID -> containerMetadata.
    * @throws IOException
    */
-  Map<Long, ContainerMetadata> getContainers(int limit) throws IOException;
+  Map<Long, ContainerMetadata> getContainers(int limit, long prevContainer)
+      throws IOException;
 
   /**
    * Delete an entry in the container DB.
@@ -98,7 +104,6 @@ public interface ContainerDBServiceProvider {
   /**
    * Get iterator to the entire container DB.
    * @return TableIterator
-   * @throws IOException exception
    */
-  TableIterator getContainerTableIterator() throws IOException;
+  TableIterator getContainerTableIterator();
 }

hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ContainerDBServiceProviderImpl.java

@@ -128,8 +128,7 @@ public class ContainerDBServiceProviderImpl
   }
 
   /**
-   * Use the DB's prefix seek iterator to start the scan from the given
-   * container ID prefix.
+   * Get key prefixes for the given container ID.
    *
    * @param containerId the given containerId.
    * @return Map of (Key-Prefix,Count of Keys).
@@ -137,14 +136,57 @@ public class ContainerDBServiceProviderImpl
   @Override
   public Map<ContainerKeyPrefix, Integer> getKeyPrefixesForContainer(
       long containerId) throws IOException {
+    // set the default startKeyPrefix to empty string
+    return getKeyPrefixesForContainer(containerId, StringUtils.EMPTY);
+  }
+
+  /**
+   * Use the DB's prefix seek iterator to start the scan from the given
+   * container ID and prev key prefix. The prev key prefix is skipped from
+   * the result.
+   *
+   * @param containerId the given containerId.
+   * @param prevKeyPrefix the given key prefix to start the scan from.
+   * @return Map of (Key-Prefix,Count of Keys).
+   */
+  @Override
+  public Map<ContainerKeyPrefix, Integer> getKeyPrefixesForContainer(
+      long containerId, String prevKeyPrefix) throws IOException {
 
     Map<ContainerKeyPrefix, Integer> prefixes = new LinkedHashMap<>();
     TableIterator<ContainerKeyPrefix, ? extends KeyValue<ContainerKeyPrefix,
         Integer>> containerIterator = containerKeyTable.iterator();
-    containerIterator.seek(new ContainerKeyPrefix(containerId));
+    ContainerKeyPrefix seekKey;
+    boolean skipPrevKey = false;
+    if (StringUtils.isNotBlank(prevKeyPrefix)) {
+      skipPrevKey = true;
+      seekKey = new ContainerKeyPrefix(containerId, prevKeyPrefix);
+    } else {
+      seekKey = new ContainerKeyPrefix(containerId);
+    }
+    KeyValue<ContainerKeyPrefix, Integer> seekKeyValue =
+        containerIterator.seek(seekKey);
+
+    // check if RocksDB was able to seek correctly to the given key prefix
+    // if not, then return empty result
+    // In case of an empty prevKeyPrefix, all the keys in the container are
+    // returned
+    if (seekKeyValue == null ||
+        (StringUtils.isNotBlank(prevKeyPrefix) &&
+            !seekKeyValue.getKey().getKeyPrefix().equals(prevKeyPrefix))) {
+      return prefixes;
+    }
+
     while (containerIterator.hasNext()) {
       KeyValue<ContainerKeyPrefix, Integer> keyValue = containerIterator.next();
       ContainerKeyPrefix containerKeyPrefix = keyValue.getKey();
+
+      // skip the prev key if prev key is present
+      if (skipPrevKey &&
+          containerKeyPrefix.getKeyPrefix().equals(prevKeyPrefix)) {
+        continue;
+      }
+
       // The prefix seek only guarantees that the iterator's head will be
       // positioned at the first prefix match. We still have to check the key
       // prefix.
@@ -164,36 +206,46 @@ public class ContainerDBServiceProviderImpl
     return prefixes;
   }
 
-  /**
-   * Get all the containers.
-   *
-   * @return Map of containerID -> containerMetadata.
-   * @throws IOException
-   */
-  @Override
-  public Map<Long, ContainerMetadata> getContainers() throws IOException {
-    // Set a negative limit to get all the containers.
-    return getContainers(-1);
-  }
-
   /**
    * Iterate the DB to construct a Map of containerID -> containerMetadata
-   * only for the given limit.
+   * only for the given limit from the given start key. The start containerID
+   * is skipped from the result.
    *
    * Return all the containers if limit < 0.
    *
+   * @param limit No of containers to get.
+   * @param prevContainer containerID after which the
+   *                      list of containers are scanned.
    * @return Map of containerID -> containerMetadata.
    * @throws IOException
    */
   @Override
-  public Map<Long, ContainerMetadata> getContainers(int limit)
+  public Map<Long, ContainerMetadata> getContainers(int limit,
+                                                    long prevContainer)
       throws IOException {
     Map<Long, ContainerMetadata> containers = new LinkedHashMap<>();
     TableIterator<ContainerKeyPrefix, ? extends KeyValue<ContainerKeyPrefix,
         Integer>> containerIterator = containerKeyTable.iterator();
+    ContainerKeyPrefix seekKey;
+    if (prevContainer > 0L) {
+      seekKey = new ContainerKeyPrefix(prevContainer);
+      KeyValue<ContainerKeyPrefix,
+          Integer> seekKeyValue = containerIterator.seek(seekKey);
+      // Check if RocksDB was able to correctly seek to the given
+      // prevContainer containerId. If not, then return empty result
+      if (seekKeyValue != null &&
+          seekKeyValue.getKey().getContainerId() != prevContainer) {
+        return containers;
+      } else {
+        // seek to the prevContainer+1 containerID to start scan
+        seekKey = new ContainerKeyPrefix(prevContainer + 1);
+        containerIterator.seek(seekKey);
+      }
+    }
     while (containerIterator.hasNext()) {
       KeyValue<ContainerKeyPrefix, Integer> keyValue = containerIterator.next();
-      Long containerID = keyValue.getKey().getContainerId();
+      ContainerKeyPrefix containerKeyPrefix = keyValue.getKey();
+      Long containerID = containerKeyPrefix.getContainerId();
       Integer numberOfKeys = keyValue.getValue();
 
       // break the loop if limit has been reached
@@ -220,7 +272,7 @@ public class ContainerDBServiceProviderImpl
   }
 
   @Override
-  public TableIterator getContainerTableIterator() throws IOException {
+  public TableIterator getContainerTableIterator() {
     return containerKeyTable.iterator();
   }
 }

hadoop-ozone/ozone-recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerKeyService.java

@@ -36,6 +36,7 @@ import java.util.Map;
 
 import javax.ws.rs.core.Response;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
@@ -132,7 +133,7 @@ public class TestContainerKeyService extends AbstractOMMetadataManagerTest {
     OmKeyLocationInfoGroup omKeyLocationInfoGroup = new
         OmKeyLocationInfoGroup(0, omKeyLocationInfoList);
 
-    //key = key_one, Blocks = [ {CID = 1, LID = 1}, {CID = 2, LID = 1} ]
+    //key = key_one, Blocks = [ {CID = 1, LID = 101}, {CID = 2, LID = 102} ]
     writeDataToOm(omMetadataManager,
         "key_one", "bucketOne", "sampleVol",
         Collections.singletonList(omKeyLocationInfoGroup));
@@ -156,7 +157,7 @@ public class TestContainerKeyService extends AbstractOMMetadataManagerTest {
     infoGroups.add(new OmKeyLocationInfoGroup(1,
         omKeyLocationInfoListNew));
 
-    //key = key_two, Blocks = [ {CID = 1, LID = 2}, {CID = 1, LID = 3} ]
+    //key = key_two, Blocks = [ {CID = 1, LID = 103}, {CID = 1, LID = 104} ]
     writeDataToOm(omMetadataManager,
         "key_two", "bucketOne", "sampleVol", infoGroups);
 
@@ -201,46 +202,83 @@ public class TestContainerKeyService extends AbstractOMMetadataManagerTest {
   @Test
   public void testGetKeysForContainer() {
 
-    Response response = containerKeyService.getKeysForContainer(1L, -1);
+    Response response = containerKeyService.getKeysForContainer(1L, -1, "");
 
     Collection<KeyMetadata> keyMetadataList =
         (Collection<KeyMetadata>) response.getEntity();
-    assertEquals(keyMetadataList.size(), 2);
+    assertEquals(2, keyMetadataList.size());
 
     Iterator<KeyMetadata> iterator = keyMetadataList.iterator();
 
     KeyMetadata keyMetadata = iterator.next();
-    assertEquals(keyMetadata.getKey(), "key_one");
-    assertEquals(keyMetadata.getVersions().size(), 1);
-    assertEquals(keyMetadata.getBlockIds().size(), 1);
+    assertEquals("key_one", keyMetadata.getKey());
+    assertEquals(1, keyMetadata.getVersions().size());
+    assertEquals(1, keyMetadata.getBlockIds().size());
     Map<Long, List<KeyMetadata.ContainerBlockMetadata>> blockIds =
         keyMetadata.getBlockIds();
-    assertEquals(blockIds.get(0L).iterator().next().getLocalID(), 101);
+    assertEquals(101, blockIds.get(0L).iterator().next().getLocalID());
 
     keyMetadata = iterator.next();
-    assertEquals(keyMetadata.getKey(), "key_two");
-    assertEquals(keyMetadata.getVersions().size(), 2);
+    assertEquals("key_two", keyMetadata.getKey());
+    assertEquals(2, keyMetadata.getVersions().size());
     assertTrue(keyMetadata.getVersions().contains(0L) && keyMetadata
         .getVersions().contains(1L));
-    assertEquals(keyMetadata.getBlockIds().size(), 2);
+    assertEquals(2, keyMetadata.getBlockIds().size());
     blockIds = keyMetadata.getBlockIds();
-    assertEquals(blockIds.get(0L).iterator().next().getLocalID(), 103);
-    assertEquals(blockIds.get(1L).iterator().next().getLocalID(), 104);
+    assertEquals(103, blockIds.get(0L).iterator().next().getLocalID());
+    assertEquals(104, blockIds.get(1L).iterator().next().getLocalID());
 
-    response = containerKeyService.getKeysForContainer(3L, -1);
+    response = containerKeyService.getKeysForContainer(3L, -1, "");
     keyMetadataList = (Collection<KeyMetadata>) response.getEntity();
     assertTrue(keyMetadataList.isEmpty());
 
     // test if limit works as expected
-    response = containerKeyService.getKeysForContainer(1L, 1);
+    response = containerKeyService.getKeysForContainer(1L, 1, "");
     keyMetadataList = (Collection<KeyMetadata>) response.getEntity();
-    assertEquals(keyMetadataList.size(), 1);
+    assertEquals(1, keyMetadataList.size());
+  }
+
+  @Test
+  public void testGetKeysForContainerWithPrevKey() {
+    // test if prev-key param works as expected
+    Response response = containerKeyService.getKeysForContainer(
+        1L, -1, "/sampleVol/bucketOne/key_one");
+
+    Collection<KeyMetadata> keyMetadataList =
+        (Collection<KeyMetadata>) response.getEntity();
+    assertEquals(1, keyMetadataList.size());
+
+    Iterator<KeyMetadata> iterator = keyMetadataList.iterator();
+    KeyMetadata keyMetadata = iterator.next();
+
+    assertEquals("key_two", keyMetadata.getKey());
+    assertEquals(2, keyMetadata.getVersions().size());
+    assertEquals(2, keyMetadata.getBlockIds().size());
+
+    response = containerKeyService.getKeysForContainer(
+        1L, -1, StringUtils.EMPTY);
+    keyMetadataList = (Collection<KeyMetadata>) response.getEntity();
+    assertEquals(2, keyMetadataList.size());
+    iterator = keyMetadataList.iterator();
+    keyMetadata = iterator.next();
+    assertEquals("key_one", keyMetadata.getKey());
+
+    // test for negative cases
+    response = containerKeyService.getKeysForContainer(
+        1L, -1, "/sampleVol/bucketOne/invalid_key");
+    keyMetadataList = (Collection<KeyMetadata>) response.getEntity();
+    assertEquals(0, keyMetadataList.size());
+
+    response = containerKeyService.getKeysForContainer(
+        5L, -1, "");
+    keyMetadataList = (Collection<KeyMetadata>) response.getEntity();
+    assertEquals(0, keyMetadataList.size());
   }
 
   @Test
   public void testGetContainers() {
 
-    Response response = containerKeyService.getContainers(-1);
+    Response response = containerKeyService.getContainers(-1, 0L);
 
     List<ContainerMetadata> containers = new ArrayList<>(
         (Collection<ContainerMetadata>) response.getEntity());
@@ -248,20 +286,55 @@ public class TestContainerKeyService extends AbstractOMMetadataManagerTest {
     Iterator<ContainerMetadata> iterator = containers.iterator();
 
     ContainerMetadata containerMetadata = iterator.next();
-    assertEquals(containerMetadata.getContainerID(), 1L);
+    assertEquals(1L, containerMetadata.getContainerID());
     // Number of keys for CID:1 should be 3 because of two different versions
     // of key_two stored in CID:1
-    assertEquals(containerMetadata.getNumberOfKeys(), 3L);
+    assertEquals(3L, containerMetadata.getNumberOfKeys());
 
     containerMetadata = iterator.next();
-    assertEquals(containerMetadata.getContainerID(), 2L);
-    assertEquals(containerMetadata.getNumberOfKeys(), 2L);
+    assertEquals(2L, containerMetadata.getContainerID());
+    assertEquals(2L, containerMetadata.getNumberOfKeys());
 
     // test if limit works as expected
-    response = containerKeyService.getContainers(1);
+    response = containerKeyService.getContainers(1, 0L);
+    containers = new ArrayList<>(
+        (Collection<ContainerMetadata>) response.getEntity());
+    assertEquals(1, containers.size());
+  }
+
+  @Test
+  public void testGetContainersWithPrevKey() {
+
+    Response response = containerKeyService.getContainers(1, 1L);
+
+    List<ContainerMetadata> containers = new ArrayList<>(
+        (Collection<ContainerMetadata>) response.getEntity());
+
+    Iterator<ContainerMetadata> iterator = containers.iterator();
+
+    ContainerMetadata containerMetadata = iterator.next();
+
+    assertEquals(1, containers.size());
+    assertEquals(2L, containerMetadata.getContainerID());
+
+    response = containerKeyService.getContainers(-1, 0L);
+    containers = new ArrayList<>(
+        (Collection<ContainerMetadata>) response.getEntity());
+    assertEquals(2, containers.size());
+    iterator = containers.iterator();
+    containerMetadata = iterator.next();
+    assertEquals(1L, containerMetadata.getContainerID());
+
+    // test for negative cases
+    response = containerKeyService.getContainers(-1, 5L);
+    containers = new ArrayList<>(
+        (Collection<ContainerMetadata>) response.getEntity());
+    assertEquals(0, containers.size());
+
+    response = containerKeyService.getContainers(-1, -1L);
     containers = new ArrayList<>(
         (Collection<ContainerMetadata>) response.getEntity());
-    assertEquals(containers.size(), 1);
+    assertEquals(2, containers.size());
   }
 
   /**

hadoop-ozone/ozone-recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestContainerDBServiceProviderImpl.java

@@ -20,6 +20,7 @@ package org.apache.hadoop.ozone.recon.spi.impl;
 
 import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_DB_DIR;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 import java.io.File;
@@ -29,6 +30,7 @@ import java.util.Map;
 
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.ozone.recon.api.types.ContainerKeyPrefix;
+import org.apache.hadoop.ozone.recon.api.types.ContainerMetadata;
 import org.apache.hadoop.ozone.recon.spi.ContainerDBServiceProvider;
 import org.apache.hadoop.utils.db.DBStore;
 import org.junit.After;
@@ -204,6 +206,112 @@ public class TestContainerDBServiceProviderImpl {
     assertTrue(keyPrefixMap.get(containerKeyPrefix3) == 3);
   }
 
+  @Test
+  public void testGetKeyPrefixesForContainerWithKeyPrefix() throws Exception {
+    long containerId = System.currentTimeMillis();
+
+    String keyPrefix1 = "V3/B1/K1";
+    String keyPrefix2 = "V3/B1/K2";
+    String keyPrefix3 = "V3/B2/K1";
+
+    ContainerKeyPrefix containerKeyPrefix1 = new
+        ContainerKeyPrefix(containerId, keyPrefix1, 0);
+    containerDbServiceProvider.storeContainerKeyMapping(containerKeyPrefix1,
+        1);
+
+    ContainerKeyPrefix containerKeyPrefix2 = new ContainerKeyPrefix(
+        containerId, keyPrefix2, 0);
+    containerDbServiceProvider.storeContainerKeyMapping(containerKeyPrefix2,
+        2);
+
+    long nextContainerId = containerId + 1000L;
+    ContainerKeyPrefix containerKeyPrefix3 = new ContainerKeyPrefix(
+        nextContainerId, keyPrefix3, 0);
+    containerDbServiceProvider.storeContainerKeyMapping(containerKeyPrefix3,
+        3);
+
+    Map<ContainerKeyPrefix, Integer> keyPrefixMap =
+        containerDbServiceProvider.getKeyPrefixesForContainer(containerId,
+            keyPrefix1);
+    assertEquals(1, keyPrefixMap.size());
+    assertEquals(2, keyPrefixMap.get(containerKeyPrefix2).longValue());
+
+    keyPrefixMap = containerDbServiceProvider.getKeyPrefixesForContainer(
+        nextContainerId, keyPrefix3);
+    assertEquals(0, keyPrefixMap.size());
+
+    // test for negative cases
+    keyPrefixMap = containerDbServiceProvider.getKeyPrefixesForContainer(
+        containerId, "V3/B1/invalid");
+    assertEquals(0, keyPrefixMap.size());
+
+    keyPrefixMap = containerDbServiceProvider.getKeyPrefixesForContainer(
+        containerId, keyPrefix3);
+    assertEquals(0, keyPrefixMap.size());
+
+    keyPrefixMap = containerDbServiceProvider.getKeyPrefixesForContainer(
+        1L, "");
+    assertEquals(0, keyPrefixMap.size());
+  }
+
+  @Test
+  public void testGetContainersWithPrevKey() throws Exception {
+    long containerId = System.currentTimeMillis();
+
+    String keyPrefix1 = "V3/B1/K1";
+    String keyPrefix2 = "V3/B1/K2";
+    String keyPrefix3 = "V3/B2/K1";
+
+    ContainerKeyPrefix containerKeyPrefix1 = new
+        ContainerKeyPrefix(containerId, keyPrefix1, 0);
+    containerDbServiceProvider.storeContainerKeyMapping(containerKeyPrefix1,
+        1);
+
+    ContainerKeyPrefix containerKeyPrefix2 = new ContainerKeyPrefix(
+        containerId, keyPrefix2, 0);
+    containerDbServiceProvider.storeContainerKeyMapping(containerKeyPrefix2,
+        2);
+
+    long nextContainerId = containerId + 1000L;
+    ContainerKeyPrefix containerKeyPrefix3 = new ContainerKeyPrefix(
+        nextContainerId, keyPrefix3, 0);
+    containerDbServiceProvider.storeContainerKeyMapping(containerKeyPrefix3,
+        3);
+
+    Map<Long, ContainerMetadata> containerMap =
+        containerDbServiceProvider.getContainers(-1, 0L);
+    assertEquals(2, containerMap.size());
+
+    assertEquals(3, containerMap.get(containerId).getNumberOfKeys());
+    assertEquals(3, containerMap.get(nextContainerId).getNumberOfKeys());
+
+    // test if limit works
+    containerMap = containerDbServiceProvider.getContainers(
+        1, 0L);
+    assertEquals(1, containerMap.size());
+    assertNull(containerMap.get(nextContainerId));
+
+    // test for prev key
+    containerMap = containerDbServiceProvider.getContainers(
+        -1, containerId);
+    assertEquals(1, containerMap.size());
+    // containerId must be skipped from containerMap result
+    assertNull(containerMap.get(containerId));
+
+    containerMap = containerDbServiceProvider.getContainers(
+        -1, nextContainerId);
+    assertEquals(0, containerMap.size());
+
+    // test for negative cases
+    containerMap = containerDbServiceProvider.getContainers(
+        -1, 1L);
+    assertEquals(0, containerMap.size());
+
+    containerMap = containerDbServiceProvider.getContainers(
+        0, containerId);
+    assertEquals(0, containerMap.size());
+  }
+
   @Test
   public void testDeleteContainerMapping() throws IOException {
     long containerId = System.currentTimeMillis();