HDDS-11530. Support listMultipartUploads max uploads and markers

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java

@@ -52,6 +52,9 @@ public enum OzoneManagerVersion implements ComponentVersion {
   S3_PART_AWARE_GET(10, "OzoneManager version that supports S3 get for a specific multipart " +
                         "upload part number"),
 
+  S3_LIST_MULTIPART_UPLOADS_PAGINATION(11,
+      "OzoneManager version that supports S3 list multipart uploads API with pagination"),
+
   FUTURE_VERSION(-1, "Used internally in the client when the server side is "
       + " newer and an unknown server version has arrived to the client.");
 

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

@@ -1029,9 +1029,10 @@ public List<OzoneFileStatus> listStatus(String keyName, boolean recursive,
    *
    * @param prefix Optional string to filter for the selected keys.
    */
-  public OzoneMultipartUploadList listMultipartUploads(String prefix)
+  public OzoneMultipartUploadList listMultipartUploads(String prefix, 
+      String keyMarker, String uploadIdMarker, int maxUploads) 
       throws IOException {
-    return proxy.listMultipartUploads(volumeName, getName(), prefix);
+    return proxy.listMultipartUploads(volumeName, getName(), prefix, keyMarker, uploadIdMarker, maxUploads);
   }
 
   /**

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

@@ -26,11 +26,20 @@
 public class OzoneMultipartUploadList {
 
   private List<OzoneMultipartUpload> uploads;
+  private String nextKeyMarker;
+  private String nextUploadIdMarker;
+  private boolean isTruncated;
 
   public OzoneMultipartUploadList(
-      List<OzoneMultipartUpload> uploads) {
+      List<OzoneMultipartUpload> uploads,
+      String nextKeyMarker,
+      String nextUploadIdMarker,
+      boolean isTruncated) {
     Preconditions.checkNotNull(uploads);
     this.uploads = uploads;
+    this.nextKeyMarker = nextKeyMarker;
+    this.nextUploadIdMarker = nextUploadIdMarker;
+    this.isTruncated = isTruncated;
   }
 
   public List<OzoneMultipartUpload> getUploads() {
@@ -41,4 +50,16 @@ public void setUploads(
       List<OzoneMultipartUpload> uploads) {
     this.uploads = uploads;
   }
+
+  public String getNextKeyMarker() {
+    return nextKeyMarker;
+  }
+
+  public String getNextUploadIdMarker() {
+    return nextUploadIdMarker;
+  }
+
+  public boolean isTruncated() {
+    return isTruncated;
+  }
 }

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

@@ -666,7 +666,7 @@ OzoneMultipartUploadPartListParts listParts(String volumeName,
    * Return with the inflight multipart uploads.
    */
   OzoneMultipartUploadList listMultipartUploads(String volumename,
-      String bucketName, String prefix) throws IOException;
+      String bucketName, String prefix, String keyMarker, String uploadIdMarker, int maxUploads) throws IOException;
 
   /**
    * Get a valid Delegation Token.

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

@@ -2124,10 +2124,16 @@ public OzoneMultipartUploadPartListParts listParts(String volumeName,
 
   @Override
   public OzoneMultipartUploadList listMultipartUploads(String volumeName,
-      String bucketName, String prefix) throws IOException {
+      String bucketName, String prefix, String keyMarker, String uploadIdMarker, int maxUploads) throws IOException {
 
-    OmMultipartUploadList omMultipartUploadList =
-        ozoneManagerClient.listMultipartUploads(volumeName, bucketName, prefix);
+    OmMultipartUploadList omMultipartUploadList;
+    if (omVersion.compareTo(OzoneManagerVersion.S3_LIST_MULTIPART_UPLOADS_PAGINATION) >= 0) {
+      omMultipartUploadList = ozoneManagerClient.listMultipartUploads(volumeName, bucketName, prefix, keyMarker,
+          uploadIdMarker, maxUploads, true);
+    } else {
+      omMultipartUploadList = ozoneManagerClient.listMultipartUploads(volumeName, bucketName, prefix, keyMarker,
+          uploadIdMarker, maxUploads, false);
+    }
     List<OzoneMultipartUpload> uploads = omMultipartUploadList.getUploads()
         .stream()
         .map(upload -> new OzoneMultipartUpload(upload.getVolumeName(),
@@ -2137,7 +2143,10 @@ public OzoneMultipartUploadList listMultipartUploads(String volumeName,
             upload.getCreationTime(),
             upload.getReplicationConfig()))
         .collect(Collectors.toList());
-    OzoneMultipartUploadList result = new OzoneMultipartUploadList(uploads);
+    OzoneMultipartUploadList result = new OzoneMultipartUploadList(uploads,
+        omMultipartUploadList.getNextKeyMarker(),
+        omMultipartUploadList.getNextUploadIdMarker(),
+        omMultipartUploadList.isTruncated());
     return result;
   }
 

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/MultipartUploadKeys.java

@@ -0,0 +1,92 @@
+/*
+ * 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.om.helpers;
+
+import java.util.Set;
+
+/**
+ * This class is used to store the result of OmMetadataManager#getMultipartUploadKeys.
+ */
+public class MultipartUploadKeys {
+  private final Set<String> keys;
+  private final String nextKeyMarker;
+  private final String nextUploadIdMarker;
+  private final boolean isTruncated;
+
+  public MultipartUploadKeys(Set<String> keys, String nextKeyMarker, String nextUploadIdMarker, boolean isTruncated) {
+    this.keys = keys;
+    this.nextKeyMarker = nextKeyMarker;
+    this.nextUploadIdMarker = nextUploadIdMarker;
+    this.isTruncated = isTruncated;
+  }
+
+  public Set<String> getKeys() {
+    return keys;
+  }
+
+  public String getNextKeyMarker() {
+    return nextKeyMarker;
+  }
+
+  public String getNextUploadIdMarker() {
+    return nextUploadIdMarker;
+  }
+
+  public boolean isTruncated() {
+    return isTruncated;
+  }
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  /**
+   * Builder class for MultipartUploadKeys.
+   */
+  public static class Builder {
+    private Set<String> keys;
+    private String nextKeyMarker = "";
+    private String nextUploadIdMarker = "";
+    private boolean isTruncated;
+
+    public Builder setKeys(Set<String> keys) {
+      this.keys = keys;
+      return this;
+    }
+
+    public Builder setNextKeyMarker(String nextKeyMarker) {
+      this.nextKeyMarker = nextKeyMarker;
+      return this;
+    }
+
+    public Builder setNextUploadIdMarker(String nextUploadIdMarker) {
+      this.nextUploadIdMarker = nextUploadIdMarker;
+      return this;
+    }
+
+    public Builder setIsTruncated(boolean isTruncated) {
+      this.isTruncated = isTruncated;
+      return this;
+    }
+
+    public MultipartUploadKeys build() {
+      return new MultipartUploadKeys(keys, nextKeyMarker, nextUploadIdMarker, isTruncated);
+    }
+
+  }
+}

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartUploadList.java

@@ -25,10 +25,19 @@
 public class OmMultipartUploadList {
 
   private List<OmMultipartUpload> uploads;
+  private String nextKeyMarker;
+  private String nextUploadIdMarker;
+  private boolean isTruncated;
 
   public OmMultipartUploadList(
-      List<OmMultipartUpload> uploads) {
+      List<OmMultipartUpload> uploads,
+      String nextKeyMarker,
+      String nextUploadIdMarker,
+      boolean isTruncated) {
     this.uploads = uploads;
+    this.nextKeyMarker = nextKeyMarker;
+    this.nextUploadIdMarker = nextUploadIdMarker;
+    this.isTruncated = isTruncated;
   }
 
   public List<OmMultipartUpload> getUploads() {
@@ -40,4 +49,16 @@ public void setUploads(
     this.uploads = uploads;
   }
 
+  public String getNextKeyMarker() {
+    return nextKeyMarker;
+  }
+
+  public String getNextUploadIdMarker() {
+    return nextUploadIdMarker;
+  }
+
+  public boolean isTruncated() {
+    return isTruncated;
+  }
+
 }

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

@@ -606,9 +606,12 @@ OmMultipartUploadListParts listParts(String volumeName, String bucketName,
 
   /**
    * List in-flight uploads.
+   * withPagination is for backward compatible as older listMultipartUploads does
+   * not support pagination.
    */
   OmMultipartUploadList listMultipartUploads(String volumeName,
-      String bucketName, String prefix) throws IOException;
+      String bucketName, String prefix,
+      String keyMarker, String uploadIdMarker, int maxUploads, boolean withPagination) throws IOException;
 
   /**
    * Gets s3Secret for given kerberos user.

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

@@ -1808,12 +1808,17 @@ public OmMultipartUploadListParts listParts(String volumeName,
   @Override
   public OmMultipartUploadList listMultipartUploads(String volumeName,
       String bucketName,
-      String prefix) throws IOException {
+      String prefix,
+      String keyMarker, String uploadIdMarker, int maxUploads, boolean withPagination) throws IOException {
     ListMultipartUploadsRequest request = ListMultipartUploadsRequest
         .newBuilder()
         .setVolume(volumeName)
         .setBucket(bucketName)
         .setPrefix(prefix == null ? "" : prefix)
+        .setKeyMarker(keyMarker == null ? "" : keyMarker)
+        .setUploadIdMarker(uploadIdMarker == null ? "" : uploadIdMarker)
+        .setMaxUploads(maxUploads)
+        .setWithPagination(withPagination)
         .build();
 
     OMRequest omRequest = createOMRequest(Type.ListMultipartUploads)
@@ -1837,7 +1842,10 @@ public OmMultipartUploadList listMultipartUploads(String volumeName,
             ))
             .collect(Collectors.toList());
 
-    OmMultipartUploadList response = new OmMultipartUploadList(uploadList);
+    OmMultipartUploadList response = new OmMultipartUploadList(uploadList,
+        listMultipartUploadsResponse.getNextKeyMarker(),
+        listMultipartUploadsResponse.getNextUploadIdMarker(),
+        listMultipartUploadsResponse.getIsTruncated());
 
     return response;
   }

hadoop-ozone/dist/src/main/smoketest/s3/MultipartUpload.robot

@@ -318,12 +318,58 @@ Test Multipart Upload Put With Copy and range with IfModifiedSince
                         Compare files           /tmp/10mb        /tmp/part-result
 
 Test Multipart Upload list
-    ${uploadID1} =      Initiate MPU    ${BUCKET}    ${PREFIX}/listtest/key1
-    ${uploadID2} =      Initiate MPU    ${BUCKET}    ${PREFIX}/listtest/key2
-
-    ${result} =         Execute AWSS3APICli     list-multipart-uploads --bucket ${BUCKET} --prefix ${PREFIX}/listtest
-                        Should contain          ${result}    ${uploadID1}
-                        Should contain          ${result}    ${uploadID2}
-
-    ${count} =          Execute and checkrc      echo '${result}' | jq -r '.Uploads | length'  0
-                        Should Be Equal          ${count}     2
+    # Create 25 multipart uploads to test pagination
+    ${uploadIDs}=    Create List
+    FOR    ${index}    IN RANGE    25
+        ${key}=    Set Variable    ${PREFIX}/listtest/key-${index}
+        ${uploadID}=    Initiate MPU    ${BUCKET}    ${key}
+        Append To List    ${uploadIDs}    ${uploadID}
+    END
+
+    # Test listing with max-items=10 (should get 3 pages: 10, 10, 5)
+    ${result}=    Execute AWSS3APICli    list-multipart-uploads --bucket ${BUCKET} --prefix ${PREFIX}/listtest --max-items 10
+
+    # Verify first page
+    ${count}=    Execute and checkrc    echo '${result}' | jq -r '.Uploads | length'    0
+    Should Be Equal    ${count}    10
+
+    ${hasNext}=    Execute and checkrc    echo '${result}' | jq -r 'has("NextToken")'    0
+    Should Be Equal    ${hasNext}    true
+
+    ${nextToken}=    Execute and checkrc    echo '${result}' | jq -r '.NextToken'    0
+    Should Not Be Empty    ${nextToken}
+
+    # Get second page
+    ${result}=    Execute AWSS3APICli    list-multipart-uploads --bucket ${BUCKET} --prefix ${PREFIX}/listtest --max-items 10 --starting-token ${nextToken}
+
+    # Verify second page
+    ${count}=    Execute and checkrc    echo '${result}' | jq -r '.Uploads | length'    0
+    Should Be Equal    ${count}    10
+
+    ${hasNext}=    Execute and checkrc    echo '${result}' | jq -r 'has("NextToken")'    0
+    Should Be Equal    ${hasNext}    true
+
+    ${nextToken}=    Execute and checkrc    echo '${result}' | jq -r '.NextToken'    0
+    Should Not Be Empty    ${nextToken}
+
+    # Get last page
+    ${result}=    Execute AWSS3APICli    list-multipart-uploads --bucket ${BUCKET} --prefix ${PREFIX}/listtest --max-items 10 --starting-token ${nextToken}
+
+    # Verify last page
+    ${count}=    Execute and checkrc    echo '${result}' | jq -r '.Uploads | length'    0
+    Should Be Equal    ${count}    5
+
+    ${hasNext}=    Execute and checkrc    echo '${result}' | jq -r 'has("NextToken")'    0
+    Should Be Equal    ${hasNext}    false
+
+    # Test prefix filtering
+    ${result}=    Execute AWSS3APICli    list-multipart-uploads --bucket ${BUCKET} --prefix ${PREFIX}/listtest/key-1
+    ${count}=    Execute and checkrc    echo '${result}' | jq -r '.Uploads | length'    0
+    Should Be Equal    ${count}    11    # Should match key-1, key-10 through key-19
+
+    # Cleanup
+    FOR    ${index}    IN RANGE    25
+        ${key}=    Set Variable    ${PREFIX}/listtest/key-${index}
+        ${uploadID}=    Get From List    ${uploadIDs}    ${index}
+        Abort MPU    ${BUCKET}    ${key}    ${uploadID}    0
+    END