HDDS-11530. Support listMultipartUploads max uploads and markers

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

@@ -1031,9 +1031,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

@@ -28,11 +28,23 @@
 public class OzoneMultipartUploadList {
 
   private List<OzoneMultipartUpload> uploads;
+  private String nextKeyMarker;
+  private String nextUploadIdMarker;
+  private boolean isTruncated;
+  private int maxUploads;
 
   public OzoneMultipartUploadList(
-      List<OzoneMultipartUpload> uploads) {
+      List<OzoneMultipartUpload> uploads,
+      String nextKeyMarker,
+      String nextUploadIdMarker,
+      boolean isTruncated,
+      int maxUploads) {
     Preconditions.checkNotNull(uploads);
     this.uploads = uploads;
+    this.nextKeyMarker = nextKeyMarker;
+    this.nextUploadIdMarker = nextUploadIdMarker;
+    this.isTruncated = isTruncated;
+    this.maxUploads = maxUploads;
   }
 
   public List<OzoneMultipartUpload> getUploads() {
@@ -43,4 +55,20 @@ public void setUploads(
       List<OzoneMultipartUpload> uploads) {
     this.uploads = uploads;
   }
+
+  public String getNextKeyMarker() {
+    return nextKeyMarker;
+  }
+
+  public String getNextUploadIdMarker() {
+    return nextUploadIdMarker;
+  }
+
+  public boolean isTruncated() {
+    return isTruncated;
+  }
+
+  public int getMaxUploads() {
+    return maxUploads;
+  }
 }

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

@@ -668,7 +668,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

@@ -2126,10 +2126,10 @@ 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);
+        ozoneManagerClient.listMultipartUploads(volumeName, bucketName, prefix, keyMarker, uploadIdMarker, maxUploads);
     List<OzoneMultipartUpload> uploads = omMultipartUploadList.getUploads()
         .stream()
         .map(upload -> new OzoneMultipartUpload(upload.getVolumeName(),
@@ -2139,7 +2139,11 @@ 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(),
+        omMultipartUploadList.getUploads().size());
     return result;
   }
 

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

@@ -26,10 +26,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() {
@@ -41,4 +50,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

@@ -610,7 +610,7 @@ OmMultipartUploadListParts listParts(String volumeName, String bucketName,
    * List in-flight uploads.
    */
   OmMultipartUploadList listMultipartUploads(String volumeName,
-      String bucketName, String prefix) throws IOException;
+      String bucketName, String prefix, String keyMarker, String uploadIdMarker, int maxUploads) throws IOException;
 
   /**
    * Gets s3Secret for given kerberos user.

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

@@ -1810,12 +1810,15 @@ 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) throws IOException {
     ListMultipartUploadsRequest request = ListMultipartUploadsRequest
         .newBuilder()
         .setVolume(volumeName)
         .setBucket(bucketName)
         .setPrefix(prefix == null ? "" : prefix)
+        .setKeyMarker(keyMarker == null ? "" : keyMarker)
+        .setUploadIdMarker(uploadIdMarker == null ? "" : uploadIdMarker)
+        .setMaxUploads(maxUploads)
         .build();
 
     OMRequest omRequest = createOMRequest(Type.ListMultipartUploads)
@@ -1839,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/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientMultipartUploadWithFSO.java

@@ -20,6 +20,8 @@
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.stream.Collectors;
 
 import javax.xml.bind.DatatypeConverter;
 import org.apache.commons.codec.digest.DigestUtils;
@@ -81,6 +83,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.UUID;
 
@@ -97,6 +100,9 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * This test verifies all the S3 multipart client apis - prefix layout.
  */
@@ -114,6 +120,8 @@ public class TestOzoneClientMultipartUploadWithFSO {
   private OzoneVolume volume;
   private OzoneBucket bucket;
 
+  private static final Logger LOG = LoggerFactory.getLogger(TestOzoneClientMultipartUploadWithFSO.class);
+
   /**
    * Create a MiniOzoneCluster for testing.
    * <p>
@@ -810,15 +818,15 @@ public void testListMultipartUpload() throws Exception {
     uploadPart(bucket, key2, uploadID2, 1, "data".getBytes(UTF_8));
     uploadPart(bucket, key3, uploadID3, 1, "data".getBytes(UTF_8));
 
-    OzoneMultipartUploadList listMPUs = bucket.listMultipartUploads("dir1");
+    OzoneMultipartUploadList listMPUs = bucket.listMultipartUploads("dir1", "", "", 1000);
     assertEquals(3, listMPUs.getUploads().size());
     List<String> expectedList = new ArrayList<>(keys);
     for (OzoneMultipartUpload mpu : listMPUs.getUploads()) {
       expectedList.remove(mpu.getKeyName());
     }
     assertEquals(0, expectedList.size());
 
-    listMPUs = bucket.listMultipartUploads("dir1/dir2");
+    listMPUs = bucket.listMultipartUploads("dir1/dir2", "", "", 1000);
     assertEquals(2, listMPUs.getUploads().size());
     expectedList = new ArrayList<>();
     expectedList.add(key2);
@@ -828,7 +836,7 @@ public void testListMultipartUpload() throws Exception {
     }
     assertEquals(0, expectedList.size());
 
-    listMPUs = bucket.listMultipartUploads("dir1/dir2/dir3");
+    listMPUs = bucket.listMultipartUploads("dir1/dir2/dir3", "", "", 1000);
     assertEquals(1, listMPUs.getUploads().size());
     expectedList = new ArrayList<>();
     expectedList.add(key3);
@@ -838,7 +846,7 @@ public void testListMultipartUpload() throws Exception {
     assertEquals(0, expectedList.size());
 
     // partial key
-    listMPUs = bucket.listMultipartUploads("d");
+    listMPUs = bucket.listMultipartUploads("d", "", "", 1000);
     assertEquals(3, listMPUs.getUploads().size());
     expectedList = new ArrayList<>(keys);
     for (OzoneMultipartUpload mpu : listMPUs.getUploads()) {
@@ -847,7 +855,7 @@ public void testListMultipartUpload() throws Exception {
     assertEquals(0, expectedList.size());
 
     // partial key
-    listMPUs = bucket.listMultipartUploads("");
+    listMPUs = bucket.listMultipartUploads("", "", "", 1000);
     assertEquals(3, listMPUs.getUploads().size());
     expectedList = new ArrayList<>(keys);
     for (OzoneMultipartUpload mpu : listMPUs.getUploads()) {
@@ -856,6 +864,70 @@ public void testListMultipartUpload() throws Exception {
     assertEquals(0, expectedList.size());
   }
 
+  @Test
+  public void testListMultipartUploadsPagination() throws Exception {
+    int numOfKeys = 25;
+    List<String> keys = new ArrayList<>();
+    Map<String, String> keyToUploadId = new HashMap<>();
+
+    // Generate keys
+    for (int i = 0; i < numOfKeys; i++) {
+      StringBuilder key = new StringBuilder();
+      int depth = 1 + i % 3; // Creates varying depth (1-3 levels)
+      for (int j = 0; j < depth; j++) {
+        key.append("dir").append(j + 1).append("/");
+      }
+      key.append("file").append(i);
+      keys.add(key.toString());
+    }
+
+    for (String key : keys) {
+      String uploadId = initiateMultipartUploadWithAsserts(bucket, key, RATIS, ONE);
+      keyToUploadId.put(key, uploadId);
+      uploadPart(bucket, key, uploadId, 1, "data".getBytes(UTF_8));
+    }
+
+    // Test full pagination process
+    final int maxUploads = 10;
+    final int expectedTruncated = 2;
+    int truncatedCount = 0;
+    String keyMarker = "";
+    String uploadIdMarker = "";
+    Set<String> retrievedKeys = new HashSet<>();
+    boolean hasMore = true;
+
+    while (hasMore) {
+      OzoneMultipartUploadList result = bucket.listMultipartUploads(
+          "dir", keyMarker, uploadIdMarker, maxUploads);
+
+      assertThat(result.getUploads())
+          .as("Number of uploads should not exceed maxUploads")
+          .hasSizeLessThanOrEqualTo(maxUploads);
+
+      assertEquals(result.getUploads().size(), result.getMaxUploads());
+
+      for (OzoneMultipartUpload upload : result.getUploads()) {
+        String key = upload.getKeyName();
+        retrievedKeys.add(key);
+
+        assertEquals(keyToUploadId.get(key), upload.getUploadId());
+      }
+
+      // Update markers for next iteration
+      keyMarker = result.getNextKeyMarker();
+      uploadIdMarker = result.getNextUploadIdMarker();
+      hasMore = result.isTruncated();
+
+      truncatedCount += result.isTruncated() ? 1 : 0;
+    }
+
+    assertEquals(keys.size(), retrievedKeys.size());
+    assertEquals(expectedTruncated, truncatedCount);
+    assertThat(retrievedKeys.stream().sorted().collect(Collectors.toList()))
+        .as("Retrieved keys should match expected keys in order")
+        .isEqualTo(keys.stream().sorted().collect(Collectors.toList()));
+  }
+
   @Test
   void testGetAllPartsWhenZeroPartNumber() throws Exception {
     String parentDir = "a/b/c/d/e/f/";

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

@@ -1759,11 +1759,16 @@ message ListMultipartUploadsRequest {
     required string volume = 1;
     required string bucket = 2;
     required string prefix = 3;
+    optional string keyMarker = 4;
+    optional string uploadIdMarker = 5;
+    optional int32 maxUploads = 6;
 }
 
 message ListMultipartUploadsResponse {
     optional bool isTruncated = 1;
     repeated MultipartUploadInfo uploadsList = 2;
+    optional string nextKeyMarker = 3;
+    optional string nextUploadIdMarker = 4;
 }
 
 message MultipartUploadInfo {