HDDS-8802. Added pagination support for ListSnapshotDiff jobs

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

@@ -24,6 +24,7 @@
 import java.util.List;
 import java.util.NoSuchElementException;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -46,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.SnapshotDiffResponse;
 import org.apache.hadoop.security.UserGroupInformation;
 
@@ -748,19 +750,75 @@ public CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
 
   /**
    * Get a list of the SnapshotDiff jobs for a bucket based on the JobStatus.
-   * @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 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 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)
+  public Iterator<OzoneSnapshotDiff> listSnapshotDiffJobs(String volumeName,
+                                                          String bucketName,
+                                                          String jobStatus,
+                                                          boolean listAll,
+                                                          String prevSnapshotDiffJob)
       throws IOException {
-    return proxy.listSnapshotDiffJobs(volumeName,
-        bucketName, jobStatus, listAll);
+    return new SnapshotDiffJobIterator(volumeName, bucketName, jobStatus, listAll, 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 listAll;
+    private String lastSnapshotDiffJob;
+    private Iterator<OzoneSnapshotDiff> currentIterator;
+
+    private SnapshotDiffJobIterator(String volumeName,
+                                    String bucketName,
+                                    String jobStatus,
+                                    boolean listAll,
+                                    String prevSnapshotDiffJob) throws IOException {
+      this.volumeName = volumeName;
+      this.bucketName = bucketName;
+      this.jobStatus = jobStatus;
+      this.listAll = listAll;
+      // 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, listAll, 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

@@ -39,7 +39,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;
@@ -67,6 +66,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.SnapshotDiffResponse;
 import org.apache.hadoop.security.KerberosInfo;
 import org.apache.hadoop.security.token.Token;
@@ -1192,13 +1192,17 @@ CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
    * @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 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)
+  ListSnapshotDiffJobResponse listSnapshotDiffJobs(String volumeName,
+                                                   String bucketName,
+                                                   String jobStatus,
+                                                   boolean listAll,
+                                                   String prevSnapshotDiffJob,
+                                                   int maxListResult)
       throws IOException;
 
   /**

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

@@ -74,7 +74,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;
@@ -141,6 +140,7 @@
 import org.apache.hadoop.ozone.security.acl.OzoneAclConfig;
 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.SnapshotDiffResponse;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
@@ -1088,20 +1088,19 @@ public CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
   }
 
   @Override
-  public List<OzoneSnapshotDiff> listSnapshotDiffJobs(String volumeName,
-                                                    String bucketName,
-                                                    String jobStatus,
-                                                    boolean listAll)
+  public ListSnapshotDiffJobResponse listSnapshotDiffJobs(String volumeName,
+                                                          String bucketName,
+                                                          String jobStatus,
+                                                          boolean listAll,
+                                                          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, listAll, prevSnapshotDiffJob,
+        maxListResult);
   }
 
   /**

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

@@ -56,7 +56,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;
@@ -70,6 +69,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.SnapshotDiffResponse;
 import org.apache.hadoop.ozone.upgrade.UpgradeFinalizer.StatusAndMessages;
 import org.apache.hadoop.security.KerberosInfo;
@@ -803,13 +803,18 @@ 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 listAll Option to specify whether to list all jobs or not.
+   * @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)
+  default ListSnapshotDiffJobResponse listSnapshotDiffJobs(String volumeName,
+                                                           String bucketName,
+                                                           String jobStatus,
+                                                           boolean listAll,
+                                                           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

@@ -222,6 +222,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.SnapshotDiffReportOzone;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.JobStatus;
@@ -1430,10 +1431,12 @@ public CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
    * {@inheritDoc}
    */
   @Override
-  public List<SnapshotDiffJob> listSnapshotDiffJobs(String volumeName,
-                                                    String bucketName,
-                                                    String jobStatus,
-                                                    boolean listAll)
+  public ListSnapshotDiffJobResponse listSnapshotDiffJobs(String volumeName,
+                                                          String bucketName,
+                                                          String jobStatus,
+                                                          boolean listAll,
+                                                          String prevSnapshotDiffJob,
+                                                          int maxListResult)
       throws IOException {
     final OzoneManagerProtocolProtos
         .ListSnapshotDiffJobRequest.Builder requestBuilder =
@@ -1442,17 +1445,24 @@ public List<SnapshotDiffJob> listSnapshotDiffJobs(String volumeName,
             .setVolumeName(volumeName)
             .setBucketName(bucketName)
             .setJobStatus(jobStatus)
-            .setListAll(listAll);
+            .setListAll(listAll)
+            .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,52 @@
+/*
+ * 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 org.apache.hadoop.ozone.om.helpers.SnapshotDiffJob;
+
+import java.util.List;
+
+/**
+ * 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 + '\'' +
+        '}';
+  }
+}

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestOmSnapshot.java

@@ -79,6 +79,7 @@
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.apache.hadoop.ozone.security.acl.OzoneObjInfo;
 import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse;
+import org.apache.hadoop.ozone.client.OzoneSnapshotDiff;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffReportOzone;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse;
 import org.apache.hadoop.ozone.upgrade.UpgradeFinalizer;
@@ -140,6 +141,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -1736,8 +1738,7 @@ public void testListSnapshotDiffWithInvalidParameters()
         " or bucket name " + bucket + " doesn't exist";
 
     Exception volBucketEx = assertThrows(OMException.class,
-        () -> store.listSnapshotDiffJobs(volume, bucket,
-            "", true));
+        () -> store.listSnapshotDiffJobs(volume, bucket, "", true, null));
     assertEquals(volBucketErrorMessage,
         volBucketEx.getMessage());
 
@@ -1747,12 +1748,11 @@ public void testListSnapshotDiffWithInvalidParameters()
     ozVolume.createBucket(bucket);
 
     assertDoesNotThrow(() ->
-        store.listSnapshotDiffJobs(volume, bucket, "", true));
+        store.listSnapshotDiffJobs(volume, bucket, "", true, null));
 
     // There are no snapshots, response should be empty.
-    assertTrue(store
-        .listSnapshotDiffJobs(volume, bucket,
-            "", true).isEmpty());
+    Iterator<OzoneSnapshotDiff> iterator = store.listSnapshotDiffJobs(volume, bucket, "", true, null);
+    assertFalse(iterator.hasNext());
 
     OzoneBucket ozBucket = ozVolume.getBucket(bucket);
     // Create keys and take snapshots.
@@ -1773,8 +1773,7 @@ public void testListSnapshotDiffWithInvalidParameters()
     String statusErrorMessage = "Invalid job status: " + invalidStatus;
 
     OMException statusEx = assertThrows(OMException.class,
-        () -> store.listSnapshotDiffJobs(volume, bucket,
-            invalidStatus, false));
+        () -> store.listSnapshotDiffJobs(volume, bucket, invalidStatus, false, null));
     assertEquals(statusErrorMessage, statusEx.getMessage());
   }
 

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -1872,6 +1872,8 @@ message ListSnapshotDiffJobRequest {
   required string bucketName = 2;
   optional string jobStatus = 3;
   optional bool listAll = 4;
+  optional string prevSnapshotDiffJob = 5;
+  optional uint32 maxListResult = 6;
 }
 
 message DeleteSnapshotRequest {
@@ -1997,6 +1999,7 @@ message CancelSnapshotDiffResponse {
 
 message ListSnapshotDiffJobResponse {
   repeated SnapshotDiffJobProto snapshotDiffJob = 1;
+  optional string lastSnapshotDiffJob = 2;
 }
 
 message DeleteSnapshotResponse {