HDDS-8802. Added pagination support for ListSnapshotDiff jobs

hadoop-ozone/cli-shell/src/main/java/org/apache/hadoop/ozone/shell/snapshot/ListSnapshotDiffHandler.java

@@ -18,10 +18,11 @@
 package org.apache.hadoop.ozone.shell.snapshot;
 
 import java.io.IOException;
-import java.util.List;
+import java.util.Iterator;
 import org.apache.hadoop.ozone.client.OzoneClient;
 import org.apache.hadoop.ozone.client.OzoneSnapshotDiff;
 import org.apache.hadoop.ozone.shell.Handler;
+import org.apache.hadoop.ozone.shell.ListOptions;
 import org.apache.hadoop.ozone.shell.OzoneAddress;
 import org.apache.hadoop.ozone.shell.bucket.BucketUri;
 import picocli.CommandLine;
@@ -37,38 +38,37 @@ public class ListSnapshotDiffHandler extends Handler {
   @CommandLine.Mixin
   private BucketUri snapshotPath;
 
-  @CommandLine.Option(names = {"-s", "--status"},
+  @CommandLine.Option(names = {"--job-status"},
       description = "List jobs based on status.\n" +
       "Accepted values are: queued, in_progress, done, failed, rejected",
       defaultValue = "in_progress")
   private String jobStatus;
 
-  @CommandLine.Option(names = {"-a", "--all"},
+  @CommandLine.Option(names = {"--all-status"},
       description = "List all jobs regardless of status.",
       defaultValue = "false")
-  private boolean listAll;
+  private boolean listAllStatus;
+
+  @CommandLine.Mixin
+  private ListOptions listOptions;
 
   @Override
   protected OzoneAddress getAddress() {
     return snapshotPath.getValue();
   }
 
   @Override
-  protected void execute(OzoneClient client, OzoneAddress address)
-      throws IOException {
-
+  protected void execute(OzoneClient client, OzoneAddress address) throws IOException {
     String volumeName = snapshotPath.getValue().getVolumeName();
     String bucketName = snapshotPath.getValue().getBucketName();
 
-    List<OzoneSnapshotDiff> jobList =
-        client.getObjectStore().listSnapshotDiffJobs(
-            volumeName, bucketName, jobStatus, listAll);
+    Iterator<OzoneSnapshotDiff> iterator = client.getObjectStore()
+        .listSnapshotDiffJobs(volumeName, bucketName, jobStatus, listAllStatus, listOptions.getStartItem());
+
+    int counter = printAsJsonArray(iterator, listOptions.getLimit());
 
-    int counter = printAsJsonArray(jobList.iterator(),
-        jobList.size());
     if (isVerbose()) {
-      System.out.printf("Found : %d snapshot diff jobs for o3://%s/ %s ",
-          counter, volumeName, bucketName);
+      System.out.printf("Found : %d snapshot diff jobs for o3://%s/ %s ", counter, volumeName, bucketName);
     }
   }
 }

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

@@ -47,6 +47,7 @@
 import org.apache.hadoop.ozone.security.OzoneTokenIdentifier;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse;
+import org.apache.hadoop.ozone.snapshot.ListSnapshotDiffJobResponse;
 import org.apache.hadoop.ozone.snapshot.ListSnapshotResponse;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -734,16 +735,73 @@ public CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
    * @param volumeName Name of the volume to which the snapshotted bucket belong
    * @param bucketName Name of the bucket to which the snapshots belong
    * @param jobStatus JobStatus to be used to filter the snapshot diff jobs
-   * @param listAll Option to specify whether to list all jobs or not
-   * @return a list of SnapshotDiffJob objects
+   * @param listAllStatus Option to specify whether to list all jobs regardless of status
+   * @param prevSnapshotDiffJob list snapshot diff jobs after this snapshot diff job.
+   * @return an iterator of SnapshotDiffJob objects
    * @throws IOException in case there is a failure while getting a response.
    */
-  public List<OzoneSnapshotDiff> listSnapshotDiffJobs(String volumeName,
-                                                    String bucketName,
-                                                    String jobStatus,
-                                                    boolean listAll)
-      throws IOException {
-    return proxy.listSnapshotDiffJobs(volumeName,
-        bucketName, jobStatus, listAll);
+  public Iterator<OzoneSnapshotDiff> listSnapshotDiffJobs(
+      String volumeName,
+      String bucketName,
+      String jobStatus,
+      boolean listAllStatus,
+      String prevSnapshotDiffJob
+  ) throws IOException {
+    return new SnapshotDiffJobIterator(volumeName, bucketName, jobStatus, listAllStatus, prevSnapshotDiffJob);
+  }
+
+  /**
+   * An Iterator to iterate over {@link SnapshotDiffJobIterator} list.
+   */
+  private final class SnapshotDiffJobIterator implements Iterator<OzoneSnapshotDiff> {
+    private final String volumeName;
+    private final String bucketName;
+    private final String jobStatus;
+    private final boolean listAllJobs;
+    private String lastSnapshotDiffJob;
+    private Iterator<OzoneSnapshotDiff> currentIterator;
+
+    private SnapshotDiffJobIterator(
+        String volumeName,
+        String bucketName,
+        String jobStatus,
+        boolean listAllStatus,
+        String prevSnapshotDiffJob) throws IOException {
+      this.volumeName = volumeName;
+      this.bucketName = bucketName;
+      this.jobStatus = jobStatus;
+      this.listAllJobs = listAllStatus;
+      // Initialized the currentIterator and lastSnapshotDiffJob.
+      getNextListOfSnapshotDiffJobs(prevSnapshotDiffJob);
+    }
+
+    @Override
+    public boolean hasNext() {
+      if (!currentIterator.hasNext() && StringUtils.isNotEmpty(lastSnapshotDiffJob)) {
+        try {
+          // fetch the next page if continuationToken is not null.
+          getNextListOfSnapshotDiffJobs(lastSnapshotDiffJob);
+        } catch (IOException e) {
+          LOG.error("Error retrieving next batch of list for snapshot diff jobs.", e);
+        }
+      }
+      return currentIterator.hasNext();
+    }
+
+    @Override
+    public OzoneSnapshotDiff next() {
+      if (hasNext()) {
+        return currentIterator.next();
+      }
+      throw new NoSuchElementException();
+    }
+
+    private void getNextListOfSnapshotDiffJobs(String prevSnapshotDiffJob) throws IOException {
+      ListSnapshotDiffJobResponse response = proxy.listSnapshotDiffJobs(volumeName, bucketName, jobStatus, listAllJobs,
+          prevSnapshotDiffJob, listCacheSize);
+      this.currentIterator =
+          response.getSnapshotDiffJobs().stream().map(OzoneSnapshotDiff::fromSnapshotDiffJob).iterator();
+      this.lastSnapshotDiffJob = response.getLastSnapshotDiffJob();
+    }
   }
 }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -38,7 +38,6 @@
 import org.apache.hadoop.ozone.client.OzoneMultipartUploadList;
 import org.apache.hadoop.ozone.client.OzoneMultipartUploadPartListParts;
 import org.apache.hadoop.ozone.client.OzoneSnapshot;
-import org.apache.hadoop.ozone.client.OzoneSnapshotDiff;
 import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.TenantArgs;
 import org.apache.hadoop.ozone.client.VolumeArgs;
@@ -68,6 +67,7 @@
 import org.apache.hadoop.ozone.security.OzoneTokenIdentifier;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse;
+import org.apache.hadoop.ozone.snapshot.ListSnapshotDiffJobResponse;
 import org.apache.hadoop.ozone.snapshot.ListSnapshotResponse;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse;
 import org.apache.hadoop.security.KerberosInfo;
@@ -1265,15 +1265,19 @@ CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
    * @param volumeName Name of the volume to which the snapshotted bucket belong
    * @param bucketName Name of the bucket to which the snapshots belong
    * @param jobStatus JobStatus to be used to filter the snapshot diff jobs
-   * @param listAll Option to specify whether to list all jobs or not
+   * @param listAllStatus Option to specify whether to list all jobs regardless of status
+   * @param prevSnapshotDiffJob list snapshot diff jobs after this snapshot diff job.
+   * @param maxListResult maximum entries to be returned from the startSnapshotDiffJob.
    * @return a list of SnapshotDiffJob objects
    * @throws IOException in case there is a failure while getting a response.
    */
-  List<OzoneSnapshotDiff> listSnapshotDiffJobs(String volumeName,
-                                               String bucketName,
-                                               String jobStatus,
-                                               boolean listAll)
-      throws IOException;
+  ListSnapshotDiffJobResponse listSnapshotDiffJobs(
+      String volumeName,
+      String bucketName,
+      String jobStatus,
+      boolean listAllStatus,
+      String prevSnapshotDiffJob,
+      int maxListResult) throws IOException;
 
   /**
    * Time to be set for given Ozone object. This operations updates modification

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -105,7 +105,6 @@
 import org.apache.hadoop.ozone.client.OzoneMultipartUploadList;
 import org.apache.hadoop.ozone.client.OzoneMultipartUploadPartListParts;
 import org.apache.hadoop.ozone.client.OzoneSnapshot;
-import org.apache.hadoop.ozone.client.OzoneSnapshotDiff;
 import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.TenantArgs;
 import org.apache.hadoop.ozone.client.VolumeArgs;
@@ -171,6 +170,7 @@
 import org.apache.hadoop.ozone.security.OzoneTokenIdentifier;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse;
+import org.apache.hadoop.ozone.snapshot.ListSnapshotDiffJobResponse;
 import org.apache.hadoop.ozone.snapshot.ListSnapshotResponse;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -1071,20 +1071,19 @@ public CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
   }
 
   @Override
-  public List<OzoneSnapshotDiff> listSnapshotDiffJobs(String volumeName,
-                                                    String bucketName,
-                                                    String jobStatus,
-                                                    boolean listAll)
-      throws IOException {
+  public ListSnapshotDiffJobResponse listSnapshotDiffJobs(
+      String volumeName,
+      String bucketName,
+      String jobStatus,
+      boolean listAllStatus,
+      String prevSnapshotDiffJob,
+      int maxListResult) throws IOException {
     Preconditions.checkArgument(StringUtils.isNotBlank(volumeName),
         "volume can't be null or empty.");
     Preconditions.checkArgument(StringUtils.isNotBlank(bucketName),
         "bucket can't be null or empty.");
-
-    return ozoneManagerClient.listSnapshotDiffJobs(
-        volumeName, bucketName, jobStatus, listAll).stream()
-        .map(OzoneSnapshotDiff::fromSnapshotDiffJob)
-        .collect(Collectors.toList());
+    return ozoneManagerClient.listSnapshotDiffJobs(volumeName, bucketName, jobStatus, listAllStatus,
+        prevSnapshotDiffJob, maxListResult);
   }
 
   /**

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java

@@ -57,7 +57,6 @@
 import org.apache.hadoop.ozone.om.helpers.S3VolumeContext;
 import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
 import org.apache.hadoop.ozone.om.helpers.ServiceInfoEx;
-import org.apache.hadoop.ozone.om.helpers.SnapshotDiffJob;
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
 import org.apache.hadoop.ozone.om.helpers.TenantStateList;
 import org.apache.hadoop.ozone.om.helpers.TenantUserInfoValue;
@@ -71,6 +70,7 @@
 import org.apache.hadoop.ozone.security.OzoneDelegationTokenSelector;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse;
+import org.apache.hadoop.ozone.snapshot.ListSnapshotDiffJobResponse;
 import org.apache.hadoop.ozone.snapshot.ListSnapshotResponse;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse;
 import org.apache.hadoop.ozone.upgrade.UpgradeFinalization;
@@ -849,14 +849,19 @@ default CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
    * @param volumeName Name of the volume to which the snapshotted bucket belong
    * @param bucketName Name of the bucket to which the snapshots belong
    * @param jobStatus JobStatus to be used to filter the snapshot diff jobs
+   * @param listAllStatus Option to specify whether to list all jobs regardless of status
+   * @param prevSnapshotDiffJob list snapshot diff jobs after this snapshot diff job.
+   * @param maxListResult maximum entries to be returned from the startSnapshotDiffJob.
    * @return a list of SnapshotDiffJob objects
    * @throws IOException in case there is a failure while getting a response.
    */
-  default List<SnapshotDiffJob> listSnapshotDiffJobs(String volumeName,
-                                                     String bucketName,
-                                                     String jobStatus,
-                                                     boolean listAll)
-      throws IOException {
+  default ListSnapshotDiffJobResponse listSnapshotDiffJobs(
+      String volumeName,
+      String bucketName,
+      String jobStatus,
+      boolean listAllStatus,
+      String prevSnapshotDiffJob,
+      int maxListResult) throws IOException {
     throw new UnsupportedOperationException("OzoneManager does not require " +
         "this to be implemented");
   }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -239,6 +239,7 @@
 import org.apache.hadoop.ozone.security.proto.SecurityProtos.GetDelegationTokenRequestProto;
 import org.apache.hadoop.ozone.security.proto.SecurityProtos.RenewDelegationTokenRequestProto;
 import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse;
+import org.apache.hadoop.ozone.snapshot.ListSnapshotDiffJobResponse;
 import org.apache.hadoop.ozone.snapshot.ListSnapshotResponse;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffReportOzone;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse;
@@ -1473,29 +1474,38 @@ public CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
    * {@inheritDoc}
    */
   @Override
-  public List<SnapshotDiffJob> listSnapshotDiffJobs(String volumeName,
-                                                    String bucketName,
-                                                    String jobStatus,
-                                                    boolean listAll)
-      throws IOException {
+  public ListSnapshotDiffJobResponse listSnapshotDiffJobs(
+      String volumeName,
+      String bucketName,
+      String jobStatus,
+      boolean listAllStatus,
+      String prevSnapshotDiffJob,
+      int maxListResult) throws IOException {
     final OzoneManagerProtocolProtos
         .ListSnapshotDiffJobRequest.Builder requestBuilder =
         OzoneManagerProtocolProtos
             .ListSnapshotDiffJobRequest.newBuilder()
             .setVolumeName(volumeName)
             .setBucketName(bucketName)
             .setJobStatus(jobStatus)
-            .setListAll(listAll);
+            .setListAll(listAllStatus)
+            .setMaxListResult(maxListResult);
+
+    if (prevSnapshotDiffJob != null) {
+      requestBuilder.setPrevSnapshotDiffJob(prevSnapshotDiffJob);
+    }
 
     final OMRequest omRequest = createOMRequest(Type.ListSnapshotDiffJobs)
         .setListSnapshotDiffJobRequest(requestBuilder)
         .build();
     final OMResponse omResponse = submitRequest(omRequest);
     handleError(omResponse);
-    return omResponse.getListSnapshotDiffJobResponse()
-        .getSnapshotDiffJobList().stream()
+    OzoneManagerProtocolProtos.ListSnapshotDiffJobResponse response = omResponse.getListSnapshotDiffJobResponse();
+    List<SnapshotDiffJob> ozoneSnapshotDiffs = response.getSnapshotDiffJobList().stream()
         .map(SnapshotDiffJob::getFromProtoBuf)
         .collect(Collectors.toList());
+    return new ListSnapshotDiffJobResponse(ozoneSnapshotDiffs,
+        response.hasLastSnapshotDiffJob() ? response.getLastSnapshotDiffJob() : null);
   }
 
   /**

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/snapshot/ListSnapshotDiffJobResponse.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.snapshot;
+
+import java.util.List;
+import org.apache.hadoop.ozone.om.helpers.SnapshotDiffJob;
+
+/**
+ * POJO for list snapshot diff job API.
+ */
+public final class ListSnapshotDiffJobResponse {
+  private final List<SnapshotDiffJob> snapshotDiffJobs;
+  private final String lastSnapshotDiffJob;
+
+  public ListSnapshotDiffJobResponse(List<SnapshotDiffJob> snapshotDiffJobs, String lastSnapshotDiffJob) {
+    this.snapshotDiffJobs = snapshotDiffJobs;
+    this.lastSnapshotDiffJob = lastSnapshotDiffJob;
+  }
+
+  public List<SnapshotDiffJob> getSnapshotDiffJobs() {
+    return snapshotDiffJobs;
+  }
+
+  public String getLastSnapshotDiffJob() {
+    return lastSnapshotDiffJob;
+  }
+
+  @Override
+  public String toString() {
+    return "ListSnapshotDiffJobResponse{" +
+        "snapshotDiffJobs: '" + snapshotDiffJobs + '\'' +
+        ", lastSnapshotDiffJob: '" + lastSnapshotDiffJob + '\'' +
+        '}';
+  }
+}