diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/ObjectStore.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/ObjectStore.java
index 592fb252794e..b7cd9e676ff5 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/ObjectStore.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/ObjectStore.java
@@ -578,14 +578,84 @@ public void deleteSnapshot(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   snapshots will be listed after this snapshot name
    * @return list of snapshots for volume/bucket snapshot path.
    * @throws IOException
    */
-  public List<OzoneSnapshot> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
-    return proxy.listSnapshot(volumeName, bucketName);
+  public Iterator<? extends OzoneSnapshot> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot) throws IOException {
+    return new SnapshotIterator(
+        volumeName, bucketName, snapshotPrefix, prevSnapshot);
+  }
+
+  /**
+   * An Iterator to iterate over {@link OzoneSnapshot} list.
+   */
+  private class SnapshotIterator implements Iterator<OzoneSnapshot> {
+
+    private String volumeName = null;
+    private String bucketName = null;
+    private String snapshotPrefix = null;
+
+    private Iterator<OzoneSnapshot> currentIterator;
+    private OzoneSnapshot currentValue;
+
+    /**
+     * Creates an Iterator to iterate over all snapshots after
+     * prevSnapshot of specified bucket. If prevSnapshot is null it iterates
+     * from the first snapshot. The returned snapshots match snapshot prefix.
+     * @param snapshotPrefix snapshot prefix to match
+     * @param prevSnapshot snapshots will be listed after this snapshot name
+     */
+    SnapshotIterator(String volumeName, String bucketName,
+                     String snapshotPrefix, String prevSnapshot)
+        throws IOException {
+      this.volumeName = volumeName;
+      this.bucketName = bucketName;
+      this.snapshotPrefix = snapshotPrefix;
+      this.currentValue = null;
+      this.currentIterator = getNextListOfSnapshots(prevSnapshot).iterator();
+    }
+
+    @Override
+    public boolean hasNext() {
+      // IMPORTANT: Without this logic, remote iteration will not work.
+      // Removing this will break the listSnapshot call if we try to
+      // list more than 1000 (ozone.client.list.cache ) snapshots.
+      if (!currentIterator.hasNext() && currentValue != null) {
+        try {
+          currentIterator = getNextListOfSnapshots(currentValue.getName())
+              .iterator();
+        } catch (IOException e) {
+          LOG.error("Error retrieving next batch of list results", e);
+        }
+      }
+      return currentIterator.hasNext();
+    }
+
+    @Override
+    public OzoneSnapshot next() {
+      if (hasNext()) {
+        currentValue = currentIterator.next();
+        return currentValue;
+      }
+      throw new NoSuchElementException();
+    }
+
+    /**
+     * Returns the next set of snapshot list using proxy.
+     * @param prevSnapshot previous snapshot, this will be excluded from result
+     * @return {@code List<OzoneSnapshot>}
+     */
+    private List<OzoneSnapshot> getNextListOfSnapshots(String prevSnapshot)
+        throws IOException {
+      return proxy.listSnapshot(volumeName, bucketName, snapshotPrefix,
+          prevSnapshot, listCacheSize);
+    }
   }
 
   /**
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
index ceb3fcad1c54..7567bcee5fdf 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
@@ -1045,13 +1045,17 @@ void deleteSnapshot(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   start of the list, this snapshot is excluded
+   * @param maxListResult  max numbet of snapshots to return
    * @return list of snapshots for volume/bucket snapshotpath.
    * @throws IOException
    */
-  List<OzoneSnapshot> listSnapshot(String volumeName, String bucketName)
-      throws IOException;
+  List<OzoneSnapshot> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException;
 
 
   /**
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
index 81d10381a911..302d266599a4 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
@@ -998,19 +998,24 @@ public SnapshotDiffResponse snapshotDiff(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   start of the list, this snapshot is excluded
+   * @param maxListResult  max numbet of snapshots to return
    * @return list of snapshots for volume/bucket snapshotpath.
    * @throws IOException
    */
   @Override
-  public List<OzoneSnapshot> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
+  public List<OzoneSnapshot> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
     Preconditions.checkArgument(Strings.isNotBlank(volumeName),
         "volume can't be null or empty.");
     Preconditions.checkArgument(Strings.isNotBlank(bucketName),
         "bucket can't be null or empty.");
-    return ozoneManagerClient.listSnapshot(volumeName, bucketName).stream()
+    return ozoneManagerClient.listSnapshot(volumeName, bucketName,
+            snapshotPrefix, prevSnapshot, maxListResult).stream()
         .map(snapshotInfo -> OzoneSnapshot.fromSnapshotInfo(snapshotInfo))
         .collect(Collectors.toList());
   }
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java
index 1a9f3c6a0b34..537fd456e285 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java
@@ -689,13 +689,17 @@ default void deleteSnapshot(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   start of the list, this snapshot is excluded
+   * @param maxListResult  max numbet of snapshots to return
    * @return list of snapshots for volume/bucket snapshotpath.
    * @throws IOException
    */
-  default List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
+  default List<SnapshotInfo> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
     throw new UnsupportedOperationException("OzoneManager does not require " +
         "this to be implemented");
   }
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
index 39206e509477..f7a73ee879cd 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
@@ -1171,13 +1171,23 @@ public void deleteSnapshot(String volumeName,
    * {@inheritDoc}
    */
   @Override
-  public List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
+  public List<SnapshotInfo> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
     final OzoneManagerProtocolProtos.ListSnapshotRequest.Builder
         requestBuilder =
         OzoneManagerProtocolProtos.ListSnapshotRequest.newBuilder()
             .setVolumeName(volumeName)
-            .setBucketName(bucketName);
+            .setBucketName(bucketName)
+            .setMaxListResult(maxListResult);
+
+    if (prevSnapshot != null) {
+      requestBuilder.setPrevSnapshot(prevSnapshot);
+    }
+
+    if (snapshotPrefix != null) {
+      requestBuilder.setPrefix(snapshotPrefix);
+    }
 
     final OMRequest omRequest = createOMRequest(Type.ListSnapshot)
         .setListSnapshotRequest(requestBuilder)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
index 8a6fb37027bd..6273fe05729b 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
@@ -78,6 +78,7 @@
 import org.apache.hadoop.ozone.client.OzoneKeyDetails;
 import org.apache.hadoop.ozone.client.OzoneKeyLocation;
 import org.apache.hadoop.ozone.client.OzoneMultipartUploadPartListParts;
+import org.apache.hadoop.ozone.client.OzoneSnapshot;
 import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.VolumeArgs;
 import org.apache.hadoop.ozone.client.io.OzoneInputStream;
@@ -4159,4 +4160,85 @@ private void assertBucketCount(OzoneVolume volume,
     Assert.assertEquals(expectedBucketCount, bucketCount);
   }
 
+  @Test
+  public void testListSnapshot() throws IOException {
+    String volumeA = "vol-a-" + RandomStringUtils.randomNumeric(5);
+    String volumeB = "vol-b-" + RandomStringUtils.randomNumeric(5);
+    String bucketA = "buc-a-" + RandomStringUtils.randomNumeric(5);
+    String bucketB = "buc-b-" + RandomStringUtils.randomNumeric(5);
+    store.createVolume(volumeA);
+    store.createVolume(volumeB);
+    OzoneVolume volA = store.getVolume(volumeA);
+    OzoneVolume volB = store.getVolume(volumeB);
+    volA.createBucket(bucketA);
+    volA.createBucket(bucketB);
+    volB.createBucket(bucketA);
+    volB.createBucket(bucketB);
+    String snapshotPrefixA = "snapshot-a-";
+    String snapshotPrefixB = "snapshot-b-";
+    for (int i = 0; i < 10; i++) {
+      store.createSnapshot(volumeA, bucketA,
+          snapshotPrefixA + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeA, bucketB,
+          snapshotPrefixA + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeB, bucketA,
+          snapshotPrefixA + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeB, bucketB,
+          snapshotPrefixA + i + "-" + RandomStringUtils.randomNumeric(5));
+    }
+    for (int i = 0; i < 10; i++) {
+      store.createSnapshot(volumeA, bucketA,
+          snapshotPrefixB + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeA, bucketB,
+          snapshotPrefixB + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeB, bucketA,
+          snapshotPrefixB + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeB, bucketB,
+          snapshotPrefixB + i + "-" + RandomStringUtils.randomNumeric(5));
+    }
+
+    Iterator<? extends OzoneSnapshot> snapshotIter =
+        store.listSnapshot(volumeA, bucketA, null, null);
+    int volABucketASnapshotCount = 0;
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      volABucketASnapshotCount++;
+    }
+    Assert.assertEquals(20, volABucketASnapshotCount);
+
+    snapshotIter = store.listSnapshot(volumeA, bucketB, null, null);
+    int volABucketBSnapshotCount = 0;
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      volABucketBSnapshotCount++;
+    }
+    Assert.assertEquals(20, volABucketASnapshotCount);
+
+    snapshotIter = store.listSnapshot(volumeB, bucketA, null, null);
+    int volBBucketASnapshotCount = 0;
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      volBBucketASnapshotCount++;
+    }
+    Assert.assertEquals(20, volABucketASnapshotCount);
+
+    snapshotIter = store.listSnapshot(volumeB, bucketB, null, null);
+    int volBBucketBSnapshotCount = 0;
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      volBBucketBSnapshotCount++;
+    }
+    Assert.assertEquals(20, volABucketASnapshotCount);
+
+    int volABucketASnapshotACount = 0;
+    snapshotIter = store.listSnapshot(volumeA, bucketA, snapshotPrefixA, null);
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      Assert.assertTrue(snapshot.getName().startsWith(snapshotPrefixA));
+      volABucketASnapshotACount++;
+    }
+    Assert.assertEquals(10, volABucketASnapshotACount);
+    Assert.assertFalse(snapshotIter.hasNext());
+
+  }
 }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmMetrics.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmMetrics.java
index 2fb153c4db8f..712b802df8aa 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmMetrics.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmMetrics.java
@@ -436,7 +436,8 @@ public void testSnapshotOps() throws Exception {
     writeClient.createSnapshot(volumeName, bucketName, snapshot2);
 
     // List snapshots
-    writeClient.listSnapshot(volumeName, bucketName);
+    writeClient.listSnapshot(
+        volumeName, bucketName, null, null, Integer.MAX_VALUE);
 
     omMetrics = getMetrics("OMMetrics");
     assertCounter("NumSnapshotActive", 2L, omMetrics);
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshot.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshot.java
index a052e4bb161b..19b74abeccd1 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshot.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshot.java
@@ -225,7 +225,7 @@ private static void preFinalizationChecks() throws Exception {
         store.createSnapshot(volumeName, bucketName,
             UUID.randomUUID().toString()));
     expectFailurePreFinalization(() ->
-        store.listSnapshot(volumeName, bucketName));
+        store.listSnapshot(volumeName, bucketName, null, null));
     expectFailurePreFinalization(() ->
         store.snapshotDiff(volumeName, bucketName,
             UUID.randomUUID().toString(),
diff --git a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
index 869713bf3c70..c962777148f6 100644
--- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
+++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
@@ -1697,6 +1697,9 @@ message CreateSnapshotRequest {
 message ListSnapshotRequest {
   optional string volumeName = 1;
   optional string bucketName = 2;
+  optional string prefix = 3;
+  optional string prevSnapshot = 4;
+  optional uint32 maxListResult = 5;
 }
 
 message SnapshotDiffRequest {
diff --git a/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java b/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java
index ffc5d91447e2..3dcabf088e6d 100644
--- a/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java
+++ b/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java
@@ -225,12 +225,16 @@ List<RepeatedOmKeyInfo> listTrash(String volumeName, String bucketName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   start of the list, this snapshot is excluded
+   * @param maxListResult  max numbet of snapshots to return
    * @return list of snapshot
    */
-  List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
-      throws IOException;
+  List<SnapshotInfo> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException;
 
   /**
    * Recover trash allows the user to recover the keys
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
index 70b318f72d45..1f122f6eebb2 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
@@ -1184,7 +1184,7 @@ public List<OmKeyInfo> listKeys(String volumeName, String bucketName,
     if (StringUtil.isNotBlank(keyPrefix)) {
       seekPrefix = getOzoneKey(volumeName, bucketName, keyPrefix);
     } else {
-      seekPrefix = getBucketKey(volumeName, bucketName + OM_KEY_PREFIX);
+      seekPrefix = getBucketKey(volumeName, bucketName) + OM_KEY_PREFIX;
     }
     int currentCount = 0;
 
@@ -1277,8 +1277,9 @@ public List<RepeatedOmKeyInfo> listTrash(String volumeName, String bucketName,
   }
 
   @Override
-  public List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
+  public List<SnapshotInfo> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
     if (Strings.isNullOrEmpty(volumeName)) {
       throw new OMException("Volume name is required.", VOLUME_NOT_FOUND);
     }
@@ -1293,39 +1294,69 @@ public List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
           BUCKET_NOT_FOUND);
     }
 
-    String prefix = getBucketKey(volumeName, bucketName + OM_KEY_PREFIX);
+    String prefix;
+    if (StringUtil.isNotBlank(snapshotPrefix)) {
+      prefix = getOzoneKey(volumeName, bucketName, snapshotPrefix);
+    } else {
+      prefix = getBucketKey(volumeName, bucketName + OM_KEY_PREFIX);
+    }
+
+    String seek;
+    if (StringUtil.isNotBlank(prevSnapshot)) {
+      // Seek to the specified snapshot.
+      seek = getOzoneKey(volumeName, bucketName, prevSnapshot);
+    } else {
+      // This allows us to seek directly to the first key with the right prefix.
+      seek = getOzoneKey(volumeName, bucketName,
+          StringUtil.isNotBlank(
+              snapshotPrefix) ? snapshotPrefix : OM_KEY_PREFIX);
+    }
+
     TreeMap<String, SnapshotInfo> snapshotInfoMap = new TreeMap<>();
 
-    appendSnapshotFromCacheToMap(snapshotInfoMap, prefix);
-    appendSnapshotFromDBToMap(snapshotInfoMap, prefix);
+    int count = appendSnapshotFromCacheToMap(
+        snapshotInfoMap, prefix, seek, maxListResult);
+    appendSnapshotFromDBToMap(
+        snapshotInfoMap, prefix, seek, count, maxListResult);
 
     return new ArrayList<>(snapshotInfoMap.values());
   }
 
-  private void appendSnapshotFromCacheToMap(
-      TreeMap snapshotInfoMap, String prefix) {
+  private int appendSnapshotFromCacheToMap(
+      TreeMap snapshotInfoMap, String prefix,
+      String previous, int maxListResult) {
+    int count = 0;
     Iterator<Map.Entry<CacheKey<String>, CacheValue<SnapshotInfo>>> iterator =
         snapshotInfoTable.cacheIterator();
-    while (iterator.hasNext()) {
+    while (iterator.hasNext() && count < maxListResult) {
       Map.Entry<CacheKey<String>, CacheValue<SnapshotInfo>> entry =
           iterator.next();
       String snapshotKey = entry.getKey().getCacheKey();
       SnapshotInfo snapshotInfo = entry.getValue().getCacheValue();
-      if (snapshotInfo != null && snapshotKey.startsWith(prefix)) {
+      if (snapshotInfo != null && snapshotKey.startsWith(prefix) &&
+          snapshotKey.compareTo(previous) > 0) {
         snapshotInfoMap.put(snapshotKey, snapshotInfo);
+        count++;
       }
     }
+    return count;
   }
 
-  private void appendSnapshotFromDBToMap(TreeMap snapshotInfoMap, String prefix)
+  private void appendSnapshotFromDBToMap(TreeMap snapshotInfoMap,
+                                         String prefix, String previous,
+                                         int count, int maxListResult)
       throws IOException {
     try (TableIterator<String, ? extends KeyValue<String, SnapshotInfo>>
              snapshotIter = snapshotInfoTable.iterator()) {
       KeyValue< String, SnapshotInfo> snapshotinfo;
-      snapshotIter.seek(prefix);
-      while (snapshotIter.hasNext()) {
+      snapshotIter.seek(previous);
+      while (snapshotIter.hasNext() && count < maxListResult) {
         snapshotinfo = snapshotIter.next();
-        if (snapshotinfo != null && snapshotinfo.getKey().startsWith(prefix)) {
+        if (snapshotinfo != null &&
+            snapshotinfo.getKey().compareTo(previous) == 0) {
+          continue;
+        }
+        if (snapshotinfo != null && snapshotinfo.getKey().startsWith(prefix))  {
           CacheValue<SnapshotInfo> cacheValue =
               snapshotInfoTable.getCacheValue(
                   new CacheKey<>(snapshotinfo.getKey()));
@@ -1334,6 +1365,7 @@ private void appendSnapshotFromDBToMap(TreeMap snapshotInfoMap, String prefix)
           // in cache.
           if (cacheValue == null) {
             snapshotInfoMap.put(snapshotinfo.getKey(), snapshotinfo.getValue());
+            count++;
           }
         } else {
           break;
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
index e07391bbfcb3..bfc927f53e12 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
@@ -2829,8 +2829,9 @@ public List<RepeatedOmKeyInfo> listTrash(String volumeName,
   }
 
   @Override
-  public List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
+  public List<SnapshotInfo> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
     if (isAclEnabled) {
       omMetadataReader.checkAcls(ResourceType.BUCKET, StoreType.OZONE,
           ACLType.LIST, volumeName, bucketName, null);
@@ -2840,7 +2841,8 @@ public List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
     auditMap.put(OzoneConsts.BUCKET, bucketName);
     try {
       metrics.incNumSnapshotLists();
-      return metadataManager.listSnapshot(volumeName, bucketName);
+      return metadataManager.listSnapshot(volumeName, bucketName,
+          snapshotPrefix, prevSnapshot, maxListResult);
     } catch (Exception ex) {
       metrics.incNumSnapshotListFails();
       auditSuccess = false;
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
index 6dbe04e3644b..f820ccf5956a 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
@@ -1270,7 +1270,8 @@ private OzoneManagerProtocolProtos.ListSnapshotResponse getSnapshots(
       OzoneManagerProtocolProtos.ListSnapshotRequest request)
       throws IOException {
     List<SnapshotInfo> snapshotInfos = impl.listSnapshot(
-        request.getVolumeName(), request.getBucketName());
+        request.getVolumeName(), request.getBucketName(), request.getPrefix(),
+        request.getPrevSnapshot(), request.getMaxListResult());
     List<OzoneManagerProtocolProtos.SnapshotInfo> snapshotInfoList =
         snapshotInfos.stream().map(SnapshotInfo::getProtobuf)
             .collect(Collectors.toList());
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
index 9075924d1afe..8f51143d5e95 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
@@ -686,26 +686,64 @@ public void testListSnapshot() throws Exception {
 
     OMRequestTestUtils.addVolumeToDB(vol1, omMetadataManager);
     addBucketsToCache(vol1, bucket1);
-    String snapshotName = "snapshot";
+    String prefixA = "snapshotA";
+    String prefixB = "snapshotB";
+    TreeSet<String> snapshotsASet = new TreeSet<>();
 
-    for (int i = 1; i <= 10; i++) {
+    for (int i = 1; i <= 100; i++) {
       if (i % 2 == 0) {
+        snapshotsASet.add(prefixA + i);
         OMRequestTestUtils.addSnapshotToTable(vol1, bucket1,
-            snapshotName + i, omMetadataManager);
+            prefixA + i, omMetadataManager);
       } else {
         OMRequestTestUtils.addSnapshotToTableCache(vol1, bucket1,
-            snapshotName + i, omMetadataManager);
+            prefixB + i, omMetadataManager);
       }
     }
 
     //Test listing all snapshots.
     List<SnapshotInfo> snapshotInfos = omMetadataManager.listSnapshot(vol1,
-        bucket1);
-    assertEquals(10, snapshotInfos.size());
+        bucket1, null, null, 100);
+    assertEquals(100, snapshotInfos.size());
+
+    snapshotInfos = omMetadataManager.listSnapshot(vol1,
+        bucket1, prefixA, null, 50);
+    assertEquals(50, snapshotInfos.size());
+    for (SnapshotInfo snapshotInfo : snapshotInfos) {
+      assertTrue(snapshotInfo.getName().startsWith(prefixA));
+    }
+
+    String startSnapshot = prefixA + 38;
+    snapshotInfos = omMetadataManager.listSnapshot(vol1,
+        bucket1, prefixA, startSnapshot, 50);
+    assertEquals(snapshotsASet.tailSet(startSnapshot).size() - 1,
+        snapshotInfos.size());
     for (SnapshotInfo snapshotInfo : snapshotInfos) {
-      assertTrue(snapshotInfo.getName().startsWith(snapshotName));
+      assertTrue(snapshotInfo.getName().startsWith(prefixA));
+      assertTrue(snapshotInfo.getName().compareTo(startSnapshot) > 0);
     }
 
+    startSnapshot = null;
+    TreeSet<String> expectedSnapshot = new TreeSet<>();
+    for (int i = 1; i <= 5; i++) {
+      snapshotInfos = omMetadataManager.listSnapshot(
+          vol1, bucket1, prefixA, startSnapshot, 10);
+      assertEquals(10, snapshotInfos.size());
+
+      for (SnapshotInfo snapshotInfo : snapshotInfos) {
+        expectedSnapshot.add(snapshotInfo.getName());
+        assertTrue(snapshotInfo.getName().startsWith(prefixA));
+        startSnapshot = snapshotInfo.getName();
+      }
+    }
+    assertEquals(snapshotsASet, expectedSnapshot);
+
+    // As now we have iterated all 50 snapshots, calling next time should
+    // return empty list.
+    snapshotInfos = omMetadataManager.listSnapshot(vol1, bucket1,
+        startSnapshot, prefixA, 10);
+
+    assertEquals(snapshotInfos.size(), 0);
   }
 
   @ParameterizedTest
@@ -721,7 +759,7 @@ public void testListSnapshotWithInvalidPath(String volume,
     addBucketsToCache(vol1, bucket1);
 
     OMException oe = assertThrows(OMException.class,
-        () -> omMetadataManager.listSnapshot(volume, bucket));
+        () -> omMetadataManager.listSnapshot(volume, bucket, null, null, 100));
     assertEquals(expectedResultCode, oe.getResult());
   }
 
@@ -758,14 +796,14 @@ public void testListSnapshotDoesNotListOtherBucketSnapshots()
 
     //Test listing snapshots only lists snapshots of specified bucket
     List<SnapshotInfo> snapshotInfos1 = omMetadataManager.listSnapshot(vol1,
-            bucket1);
+            bucket1, null, null, Integer.MAX_VALUE);
     assertEquals(2, snapshotInfos1.size());
     for (SnapshotInfo snapshotInfo : snapshotInfos1) {
       assertTrue(snapshotInfo.getName().startsWith(snapshotName1));
     }
 
     List<SnapshotInfo> snapshotInfos2 = omMetadataManager.listSnapshot(vol1,
-            bucket2);
+            bucket2, null, null, Integer.MAX_VALUE);
     assertEquals(5, snapshotInfos2.size());
     for (SnapshotInfo snapshotInfo : snapshotInfos2) {
       assertTrue(snapshotInfo.getName().startsWith(snapshotName2));
diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneClientAdapterImpl.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneClientAdapterImpl.java
index d0e6ab2c7781..382f1df3856b 100644
--- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneClientAdapterImpl.java
+++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneClientAdapterImpl.java
@@ -817,19 +817,24 @@ private List<FileStatusAdapter> listStatusVolume(String volumeStr,
    */
   private List<FileStatusAdapter> listStatusBucketSnapshot(
       String volumeName, String bucketName, URI uri) throws IOException {
-    List<OzoneSnapshot> snapshotList =
-        objectStore.listSnapshot(volumeName, bucketName);
 
     OzoneBucket ozoneBucket = getBucket(volumeName, bucketName, false);
     UserGroupInformation ugi =
         UserGroupInformation.createRemoteUser(ozoneBucket.getOwner());
     String owner = ugi.getShortUserName();
     String group = getGroupName(ugi);
+    List<FileStatusAdapter> res = new ArrayList<>();
 
-    return snapshotList.stream()
-        .map(ozoneSnapshot -> getFileStatusAdapterForBucketSnapshot(
-            ozoneBucket, ozoneSnapshot, uri, owner, group))
-        .collect(Collectors.toList());
+    Iterator<? extends OzoneSnapshot> snapshotIter =
+        objectStore.listSnapshot(volumeName, bucketName, null, null);
+
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot ozoneSnapshot = snapshotIter.next();
+      res.add(getFileStatusAdapterForBucketSnapshot(
+          ozoneBucket, ozoneSnapshot, uri, owner, group));
+    }
+
+    return res;
   }
 
   /**
diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java
index 69f8a2f499a1..2067f518320b 100644
--- a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java
+++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java
@@ -618,17 +618,19 @@ public String createSnapshot(String volumeName,
   }
 
   @Override
+  public List<OzoneSnapshot> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
+    return null;
+  }
+  
   public void deleteSnapshot(String volumeName,
       String bucketName, String snapshotName)
       throws IOException {
 
   }
 
-  @Override
-  public List<OzoneSnapshot> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
-    return null;
-  }
+
 
   @Override
   public SnapshotDiffResponse snapshotDiff(String volumeName,
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/snapshot/ListSnapshotHandler.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/snapshot/ListSnapshotHandler.java
index 7ad4041bef3a..8ac9471f2607 100644
--- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/snapshot/ListSnapshotHandler.java
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/snapshot/ListSnapshotHandler.java
@@ -25,7 +25,7 @@
 import picocli.CommandLine;
 
 import java.io.IOException;
-import java.util.List;
+import java.util.Iterator;
 
 /**
  * ozone sh snapshot list.
@@ -50,10 +50,9 @@ protected void execute(OzoneClient client, OzoneAddress address)
     String volumeName = snapshotPath.getValue().getVolumeName();
     String bucketName = snapshotPath.getValue().getBucketName();
 
-    List<? extends OzoneSnapshot> snapshotInfos = client.getObjectStore()
-        .listSnapshot(volumeName, bucketName);
-    int counter = printAsJsonArray(snapshotInfos.iterator(),
-        snapshotInfos.size());
+    Iterator<? extends OzoneSnapshot> snapshotInfos = client.getObjectStore()
+        .listSnapshot(volumeName, bucketName, null, null);
+    int counter = printAsJsonArray(snapshotInfos, Integer.MAX_VALUE);
     if (isVerbose()) {
       out().printf("Found : %d snapshots for o3://%s/ %s ", counter,
           volumeName, bucketName);