diff --git a/backend/azure/azure.go b/backend/azure/azure.go
index 4975984..c89054e 100644
--- a/backend/azure/azure.go
+++ b/backend/azure/azure.go
@@ -1355,25 +1355,41 @@ func (az *Azure) ListParts(ctx context.Context, input *s3.ListPartsInput) (s3res
 
 // Lists all the multipart uploads initiated with .sgwtmp/multipart prefix
 func (az *Azure) ListMultipartUploads(ctx context.Context, input *s3.ListMultipartUploadsInput) (s3response.ListMultipartUploadsResult, error) {
-	client, err := az.getContainerClient(*input.Bucket)
+	var bucket string
+	if input.Bucket != nil {
+		bucket = *input.Bucket
+	}
+
+	client, err := az.getContainerClient(bucket)
 	if err != nil {
 		return s3response.ListMultipartUploadsResult{}, err
 	}
 
-	uploads := []s3response.Upload{}
-
+	var delimiter string
+	if input.Delimiter != nil {
+		delimiter = *input.Delimiter
+	}
+	var prefix string
+	if input.Prefix != nil {
+		prefix = *input.Prefix
+	}
+	var keyMarker string
+	if input.KeyMarker != nil {
+		keyMarker = *input.KeyMarker
+	}
 	var uploadIDMarker string
 	if input.UploadIdMarker != nil {
 		uploadIDMarker = *input.UploadIdMarker
 	}
-	uploadIdMarkerFound := false
-	prefix := string(metaTmpMultipartPrefix)
+	maxUploads := int(*input.MaxUploads)
 
+	mpPrefix := string(metaTmpMultipartPrefix)
 	pager := client.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{
 		Include: container.ListBlobsInclude{Metadata: true},
-		Prefix:  &prefix,
+		Prefix:  &mpPrefix,
 	})
 
+	uploads := []s3response.Upload{}
 	for pager.More() {
 		resp, err := pager.NextPage(ctx)
 		if err != nil {
@@ -1384,10 +1400,10 @@ func (az *Azure) ListMultipartUploads(ctx context.Context, input *s3.ListMultipa
 			if !ok {
 				continue
 			}
-			if *key <= *input.KeyMarker {
+			if keyMarker != "" && *key <= keyMarker {
 				continue
 			}
-			if input.Prefix != nil && !strings.HasPrefix(*key, *input.Prefix) {
+			if prefix != "" && !strings.HasPrefix(*key, prefix) {
 				continue
 			}
 
@@ -1403,62 +1419,33 @@ func (az *Azure) ListMultipartUploads(ctx context.Context, input *s3.ListMultipa
 			})
 		}
 	}
-	maxUploads := 1000
-	if input.MaxUploads != nil {
-		maxUploads = int(*input.MaxUploads)
-	}
-	if *input.KeyMarker != "" && uploadIDMarker != "" && !uploadIdMarkerFound {
-		return s3response.ListMultipartUploadsResult{
-			Bucket:         *input.Bucket,
-			Delimiter:      *input.Delimiter,
-			KeyMarker:      *input.KeyMarker,
-			MaxUploads:     maxUploads,
-			Prefix:         *input.Prefix,
-			UploadIDMarker: *input.UploadIdMarker,
-			Uploads:        []s3response.Upload{},
-		}, nil
-	}
 
+	// Sort once: Key asc, Initiated asc
 	sort.SliceStable(uploads, func(i, j int) bool {
-		return uploads[i].Key < uploads[j].Key
+		if uploads[i].Key != uploads[j].Key {
+			return uploads[i].Key < uploads[j].Key
+		}
+		return uploads[i].Initiated.Before(uploads[j].Initiated)
 	})
 
-	if *input.KeyMarker != "" && *input.UploadIdMarker != "" {
-		// the uploads are already filtered by keymarker
-		// filter the uploads by uploadIdMarker
-		for i, upl := range uploads {
-			if upl.UploadID == uploadIDMarker {
-				uploads = uploads[i+1:]
-				break
-			}
-		}
+	result, err := backend.ListMultipartUploads(uploads, prefix, delimiter, keyMarker, uploadIDMarker, maxUploads)
+	if err != nil {
+		return s3response.ListMultipartUploadsResult{}, err
 	}
 
-	if len(uploads) <= maxUploads {
-		return s3response.ListMultipartUploadsResult{
-			Bucket:         *input.Bucket,
-			Delimiter:      *input.Delimiter,
-			KeyMarker:      *input.KeyMarker,
-			MaxUploads:     maxUploads,
-			Prefix:         *input.Prefix,
-			UploadIDMarker: *input.UploadIdMarker,
-			Uploads:        uploads,
-		}, nil
-	} else {
-		resUploads := uploads[:maxUploads]
-		return s3response.ListMultipartUploadsResult{
-			Bucket:             *input.Bucket,
-			Delimiter:          *input.Delimiter,
-			KeyMarker:          *input.KeyMarker,
-			NextKeyMarker:      resUploads[len(resUploads)-1].Key,
-			MaxUploads:         maxUploads,
-			Prefix:             *input.Prefix,
-			UploadIDMarker:     *input.UploadIdMarker,
-			NextUploadIDMarker: resUploads[len(resUploads)-1].UploadID,
-			IsTruncated:        true,
-			Uploads:            resUploads,
-		}, nil
-	}
+	return s3response.ListMultipartUploadsResult{
+		Bucket:             bucket,
+		Delimiter:          delimiter,
+		KeyMarker:          keyMarker,
+		MaxUploads:         maxUploads,
+		Prefix:             prefix,
+		NextKeyMarker:      result.NextKeyMarker,
+		NextUploadIDMarker: result.NextUploadIDMarker,
+		UploadIDMarker:     uploadIDMarker,
+		IsTruncated:        result.IsTruncated,
+		Uploads:            result.Uploads,
+		CommonPrefixes:     result.CommonPrefixes,
+	}, nil
 }
 
 // Deletes the block blob with committed/uncommitted blocks
diff --git a/backend/mp-lister.go b/backend/mp-lister.go
new file mode 100644
index 0000000..fb54869
--- /dev/null
+++ b/backend/mp-lister.go
@@ -0,0 +1,205 @@
+// Copyright 2026 Versity Software
+// This file is licensed 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 backend
+
+import (
+	"strings"
+
+	"github.com/google/uuid"
+	"github.com/versity/versitygw/s3err"
+	"github.com/versity/versitygw/s3response"
+)
+
+// ListMultipartUploads initializes a multipart upload lister and calls Run()
+func ListMultipartUploads(uploads []s3response.Upload, prefix, delimiter, keyMarker, uploadIdMarker string, maxUploads int) (*ListMultipartUploadsPage, error) {
+	lister := &MultipartUploadLister{
+		Uploads:        uploads,
+		Prefix:         prefix,
+		Delimiter:      delimiter,
+		KeyMarker:      keyMarker,
+		UploadIDMarker: uploadIdMarker,
+		MaxUploads:     maxUploads,
+	}
+
+	return lister.Run()
+}
+
+// MultipartUploadLister emits a ListMultipartUploads-compatible page from an
+// already-sorted, already prefix- and key-marker-filtered upload list.
+//
+// Assumptions about input Uploads:
+//   - Sorted by (Key asc, Initiated asc)
+//   - Filtered by Prefix
+//   - Filtered to start strictly after key-marker when key-marker was provided.
+type MultipartUploadLister struct {
+	Uploads        []s3response.Upload
+	Prefix         string
+	Delimiter      string
+	MaxUploads     int
+	KeyMarker      string
+	UploadIDMarker string
+}
+
+// ListMultipartUploadsPage is the lister output
+type ListMultipartUploadsPage struct {
+	Uploads            []s3response.Upload
+	CommonPrefixes     []s3response.CommonPrefix
+	IsTruncated        bool
+	NextKeyMarker      string
+	NextUploadIDMarker string
+}
+
+// Run validates marker constraints, then performs a single-pass list that:
+// - collapses uploads into CommonPrefixes when delimiter is set
+// - enforces MaxUploads over (Uploads + CommonPrefixes)
+// - computes truncation and next markers
+func (l *MultipartUploadLister) Run() (*ListMultipartUploadsPage, error) {
+	out := &ListMultipartUploadsPage{}
+
+	var startIndex int
+
+	// if upload-id-marker is provided without a corresponding key-marker, ignore it.
+	uploadIDMarker := l.UploadIDMarker
+	if l.KeyMarker == "" {
+		uploadIDMarker = ""
+	}
+
+	if uploadIDMarker != "" {
+		// any invalid uuid is considered as an invalid uploadIdMarker
+		_, err := uuid.Parse(uploadIDMarker)
+		if err != nil {
+			return nil, s3err.GetAPIError(s3err.ErrInvalidUploadIdMarker)
+		}
+		startIndex = l.findUploadIdMarkerIndex(uploadIDMarker)
+		if startIndex == -1 {
+			return nil, s3err.GetAPIError(s3err.ErrInvalidUploadIdMarker)
+		}
+		if startIndex >= len(l.Uploads) {
+			return out, nil
+		}
+	}
+
+	// Common prefix uniqueness tracking.
+	seenCP := make(map[string]struct{})
+
+	emitted := 0
+	var lastKey string
+
+	// emitUpload appends a new upload to out.Uplodas
+	emitUpload := func(up s3response.Upload) bool {
+		out.Uploads = append(out.Uploads, up)
+		emitted++
+		lastKey = up.Key
+		return emitted == l.MaxUploads
+	}
+	// emitCp appends a new common prefix to out.CommonPrefixes
+	emitCP := func(cpref string) bool {
+		out.CommonPrefixes = append(out.CommonPrefixes, s3response.CommonPrefix{Prefix: cpref})
+		emitted++
+		lastKey = cpref
+		return emitted == l.MaxUploads
+	}
+
+	for i, up := range l.Uploads[startIndex:] {
+		if l.Delimiter != "" {
+			// delimiter check
+			suffix := strings.TrimPrefix(up.Key, l.Prefix)
+			before, _, found := strings.Cut(suffix, l.Delimiter)
+			if found {
+				cpref := l.Prefix + before + l.Delimiter
+				if _, ok := seenCP[cpref]; !ok {
+					seenCP[cpref] = struct{}{}
+					if emitCP(cpref) {
+						out.IsTruncated = l.hasMoreAfter(i+1, seenCP)
+						if out.IsTruncated {
+							out.NextKeyMarker = lastKey
+							out.NextUploadIDMarker = up.UploadID
+						}
+						return out, nil
+					}
+				}
+				continue
+			}
+		}
+
+		if emitUpload(up) {
+			out.IsTruncated = l.hasMoreAfter(i+1, seenCP)
+			if out.IsTruncated {
+				out.NextKeyMarker = lastKey
+				out.NextUploadIDMarker = up.UploadID
+			}
+			return out, nil
+		}
+	}
+
+	return out, nil
+}
+
+// findUploadIdMarkerIndex finds the index of given uploadId marker in uploads
+// uploadIDMarker must match an upload-id among uploads with the first key after KeyMarker.
+// Since caller filtered to Key > KeyMarker and the list is sorted by key/time,
+// the first key after KeyMarker is Uploads[0].Key (if any).
+// -1 is returned if no uploadId is found
+func (l *MultipartUploadLister) findUploadIdMarkerIndex(uploadIDMarker string) int {
+	if len(l.Uploads) == 0 {
+		// key-marker provided but nothing after it => upload-id-marker can never be valid.
+		return -1
+	}
+	firstKey := l.Uploads[0].Key
+
+	// it must match an upload-id under firstKey only.
+	// If firstKey has multiple uploads, any of those IDs is valid.
+	for i, up := range l.Uploads {
+		if up.Key != firstKey {
+			// sorted by key, so we're past firstKey group
+			break
+		}
+		if up.UploadID == uploadIDMarker {
+			// the listing should start from the next index
+			// to skip the uploadId marker
+			return i + 1
+		}
+	}
+	return -1
+}
+
+// hasMoreAfter checks if there exists at least one more effective item after idx,
+// considering delimiter collapse and already-emitted common prefixes.
+func (l *MultipartUploadLister) hasMoreAfter(idx int, seenCP map[string]struct{}) bool {
+	if idx >= len(l.Uploads) {
+		return false
+	}
+	if l.Delimiter == "" {
+		// any remaining upload would be emitted
+		return true
+	}
+
+	for i := idx; i < len(l.Uploads); i++ {
+		up := l.Uploads[i]
+		suffix := strings.TrimPrefix(up.Key, l.Prefix)
+		before, _, found := strings.Cut(suffix, l.Delimiter)
+		if !found {
+			// would emit an upload
+			return true
+		}
+		cpref := l.Prefix + before + l.Delimiter
+		if _, ok := seenCP[cpref]; ok {
+			continue
+		}
+		// would emit a new common prefix
+		return true
+	}
+	return false
+}
diff --git a/backend/posix/posix.go b/backend/posix/posix.go
index 492c44f..89112f9 100644
--- a/backend/posix/posix.go
+++ b/backend/posix/posix.go
@@ -2180,11 +2180,12 @@ func (p *Posix) ListMultipartUploads(_ context.Context, mpu *s3.ListMultipartUpl
 	var lmu s3response.ListMultipartUploadsResult
 
 	bucket := *mpu.Bucket
-	var delimiter string
 
 	if !p.isBucketValid(bucket) {
 		return lmu, s3err.GetAPIError(s3err.ErrInvalidBucketName)
 	}
+
+	var delimiter string
 	if mpu.Delimiter != nil {
 		delimiter = *mpu.Delimiter
 	}
@@ -2192,6 +2193,15 @@ func (p *Posix) ListMultipartUploads(_ context.Context, mpu *s3.ListMultipartUpl
 	if mpu.Prefix != nil {
 		prefix = *mpu.Prefix
 	}
+	var keyMarker string
+	if mpu.KeyMarker != nil {
+		keyMarker = *mpu.KeyMarker
+	}
+	var uploadIDMarker string
+	if mpu.UploadIdMarker != nil {
+		uploadIDMarker = *mpu.UploadIdMarker
+	}
+	maxUploads := int(*mpu.MaxUploads)
 
 	_, err := os.Stat(bucket)
 	if errors.Is(err, fs.ErrNotExist) {
@@ -2205,17 +2215,6 @@ func (p *Posix) ListMultipartUploads(_ context.Context, mpu *s3.ListMultipartUpl
 	objs, _ := os.ReadDir(filepath.Join(bucket, MetaTmpMultipartDir))
 
 	var uploads []s3response.Upload
-	var resultUpds []s3response.Upload
-
-	var keyMarker string
-	if mpu.KeyMarker != nil {
-		keyMarker = *mpu.KeyMarker
-	}
-	var uploadIDMarker string
-	if mpu.UploadIdMarker != nil {
-		uploadIDMarker = *mpu.UploadIdMarker
-	}
-	keyMarkerInd, uploadIdMarkerFound := -1, false
 
 	for _, obj := range objs {
 		if !obj.IsDir() {
@@ -2227,7 +2226,12 @@ func (p *Posix) ListMultipartUploads(_ context.Context, mpu *s3.ListMultipartUpl
 			continue
 		}
 		objectName := string(b)
-		if mpu.Prefix != nil && !strings.HasPrefix(objectName, *mpu.Prefix) {
+		// filter by prefix
+		if prefix != "" && !strings.HasPrefix(objectName, prefix) {
+			continue
+		}
+		// filter by keyMarker
+		if keyMarker != "" && objectName <= keyMarker {
 			continue
 		}
 
@@ -2241,22 +2245,12 @@ func (p *Posix) ListMultipartUploads(_ context.Context, mpu *s3.ListMultipartUpl
 				continue
 			}
 
-			// userMetaData := make(map[string]string)
-			// upiddir := filepath.Join(bucket, metaTmpMultipartDir, obj.Name(), upid.Name())
-			// loadUserMetaData(upiddir, userMetaData)
-
 			fi, err := upid.Info()
 			if err != nil {
 				return lmu, fmt.Errorf("stat %q: %w", upid.Name(), err)
 			}
 
 			uploadID := upid.Name()
-			if !uploadIdMarkerFound && uploadIDMarker == uploadID {
-				uploadIdMarkerFound = true
-			}
-			if keyMarkerInd == -1 && objectName == keyMarker {
-				keyMarkerInd = len(uploads)
-			}
 
 			checksum, err := p.retrieveChecksums(nil, bucket, filepath.Join(MetaTmpMultipartDir, obj.Name(), uploadID))
 			if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
@@ -2274,61 +2268,31 @@ func (p *Posix) ListMultipartUploads(_ context.Context, mpu *s3.ListMultipartUpl
 		}
 	}
 
-	maxUploads := int(*mpu.MaxUploads)
-	if (uploadIDMarker != "" && !uploadIdMarkerFound) || (keyMarker != "" && keyMarkerInd == -1) {
-		return s3response.ListMultipartUploadsResult{
-			Bucket:         bucket,
-			Delimiter:      delimiter,
-			KeyMarker:      keyMarker,
-			MaxUploads:     maxUploads,
-			Prefix:         prefix,
-			UploadIDMarker: uploadIDMarker,
-			Uploads:        []s3response.Upload{},
-		}, nil
-	}
-
+	// Sort once: Key asc, Initiated asc
 	sort.SliceStable(uploads, func(i, j int) bool {
-		return uploads[i].Key < uploads[j].Key
+		if uploads[i].Key != uploads[j].Key {
+			return uploads[i].Key < uploads[j].Key
+		}
+		return uploads[i].Initiated.Before(uploads[j].Initiated)
 	})
 
-	start := 0
-	if keyMarker != "" {
-		for i, up := range uploads {
-			if up.Key == keyMarker && (uploadIDMarker == "" ||
-				up.UploadID == uploadIDMarker) {
-				// Start after the marker
-				start = i + 1
-				break
-			}
-		}
-	}
-
-	for i := start; i < len(uploads); i++ {
-		if len(resultUpds) == maxUploads {
-			return s3response.ListMultipartUploadsResult{
-				Bucket:             bucket,
-				Delimiter:          delimiter,
-				KeyMarker:          keyMarker,
-				MaxUploads:         maxUploads,
-				NextKeyMarker:      resultUpds[len(resultUpds)-1].Key,
-				NextUploadIDMarker: resultUpds[len(resultUpds)-1].UploadID,
-				IsTruncated:        true,
-				Prefix:             prefix,
-				UploadIDMarker:     uploadIDMarker,
-				Uploads:            resultUpds,
-			}, nil
-		}
-		resultUpds = append(resultUpds, uploads[i])
+	result, err := backend.ListMultipartUploads(uploads, prefix, delimiter, keyMarker, uploadIDMarker, maxUploads)
+	if err != nil {
+		return lmu, err
 	}
 
 	return s3response.ListMultipartUploadsResult{
-		Bucket:         bucket,
-		Delimiter:      delimiter,
-		KeyMarker:      keyMarker,
-		MaxUploads:     maxUploads,
-		Prefix:         prefix,
-		UploadIDMarker: uploadIDMarker,
-		Uploads:        resultUpds,
+		Bucket:             bucket,
+		Delimiter:          delimiter,
+		KeyMarker:          keyMarker,
+		MaxUploads:         maxUploads,
+		Prefix:             prefix,
+		NextKeyMarker:      result.NextKeyMarker,
+		NextUploadIDMarker: result.NextUploadIDMarker,
+		UploadIDMarker:     uploadIDMarker,
+		IsTruncated:        result.IsTruncated,
+		Uploads:            result.Uploads,
+		CommonPrefixes:     result.CommonPrefixes,
 	}, nil
 }
 
diff --git a/cmd/versitygw/test.go b/cmd/versitygw/test.go
index 30c1a92..e87363e 100644
--- a/cmd/versitygw/test.go
+++ b/cmd/versitygw/test.go
@@ -374,6 +374,9 @@ func extractIntTests() (commands []*cli.Command) {
 				if hostStyle {
 					opts = append(opts, integration.WithHostStyle())
 				}
+				if azureTests {
+					opts = append(opts, integration.WithAzureMode())
+				}
 
 				s := integration.NewS3Conf(opts...)
 				err := testFunc(s)
@@ -386,6 +389,12 @@ func extractIntTests() (commands []*cli.Command) {
 					Destination: &versioningEnabled,
 					Aliases:     []string{"vs"},
 				},
+				&cli.BoolFlag{
+					Name:        "azure-test-mode",
+					Usage:       "Skips tests that are not supported by Azure",
+					Destination: &azureTests,
+					Aliases:     []string{"azure"},
+				},
 			},
 		})
 	}
diff --git a/s3err/s3err.go b/s3err/s3err.go
index 9516d92..11847b0 100644
--- a/s3err/s3err.go
+++ b/s3err/s3err.go
@@ -177,6 +177,7 @@ const (
 	ErrTrailerHeaderNotSupported
 	ErrBadRequest
 	ErrMissingUploadId
+	ErrInvalidUploadIdMarker
 	ErrNoSuchCORSConfiguration
 	ErrCORSForbidden
 	ErrMissingCORSOrigin
@@ -796,6 +797,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "This operation does not accept partNumber without uploadId",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrInvalidUploadIdMarker: {
+		Code:           "InvalidArgument",
+		Description:    "Invalid uploadId marker",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	ErrNoSuchCORSConfiguration: {
 		Code:           "NoSuchCORSConfiguration",
 		Description:    "The CORS configuration does not exist",
diff --git a/tests/integration/ListMultipartUploads.go b/tests/integration/ListMultipartUploads.go
index a619f51..c86ac9a 100644
--- a/tests/integration/ListMultipartUploads.go
+++ b/tests/integration/ListMultipartUploads.go
@@ -17,9 +17,13 @@ package integration
 import (
 	"context"
 	"fmt"
+	"sort"
+	"strings"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/google/uuid"
 	"github.com/versity/versitygw/s3err"
 )
 
@@ -174,34 +178,6 @@ func ListMultipartUploads_exceeding_max_uploads(s *S3Conf) error {
 	})
 }
 
-func ListMultipartUploads_incorrect_next_key_marker(s *S3Conf) error {
-	testName := "ListMultipartUploads_incorrect_next_key_marker"
-	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		for i := 1; i < 6; i++ {
-			_, err := createMp(s3client, bucket, fmt.Sprintf("obj%v", i))
-			if err != nil {
-				return err
-			}
-		}
-		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-		out, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
-			Bucket:    &bucket,
-			KeyMarker: getPtr("wrong_object_key"),
-		})
-		cancel()
-		if err != nil {
-			return err
-		}
-
-		if len(out.Uploads) != 0 {
-			return fmt.Errorf("expected empty list of multipart uploads, instead got %v",
-				out.Uploads)
-		}
-
-		return nil
-	})
-}
-
 func ListMultipartUploads_ignore_upload_id_marker(s *S3Conf) error {
 	testName := "ListMultipartUploads_ignore_upload_id_marker"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -226,11 +202,378 @@ func ListMultipartUploads_ignore_upload_id_marker(s *S3Conf) error {
 		if err != nil {
 			return err
 		}
-		if ok := compareMultipartUploads(out.Uploads, uploads); !ok {
+		if !compareMultipartUploads(out.Uploads, uploads) {
 			return fmt.Errorf("expected multipart uploads to be %v, instead got %v",
 				uploads, out.Uploads)
 		}
 
+		// should ignore invalid uploaId marker
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err = s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:         &bucket,
+			UploadIdMarker: getPtr("invalid_uploadId_marker"),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+		if !compareMultipartUploads(out.Uploads, uploads) {
+			return fmt.Errorf("expected multipart uploads to be %v, instead got %v",
+				uploads, out.Uploads)
+		}
+
+		return nil
+	})
+}
+
+func ListMultipartUploads_invalid_uploadId_marker(s *S3Conf) error {
+	testName := "ListMultipartUploads_invalid_uploadId_marker"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		uploads := make([]types.MultipartUpload, 0, 5)
+		for i := range 5 {
+			out, err := createMp(s3client, bucket, fmt.Sprintf("obj-%v", i))
+			if err != nil {
+				return err
+			}
+
+			uploads = append(uploads, types.MultipartUpload{
+				UploadId:     out.UploadId,
+				Key:          out.Key,
+				StorageClass: types.StorageClassStandard,
+			})
+		}
+
+		// invalid UUID
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:         &bucket,
+			KeyMarker:      getPtr("obj-2"),
+			UploadIdMarker: getPtr("invalid_uploadId_marker"),
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidUploadIdMarker)); err != nil {
+			return err
+		}
+
+		// valid UUID, but not from the list
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:         &bucket,
+			KeyMarker:      getPtr("obj-2"),
+			UploadIdMarker: getPtr(uuid.New().String()),
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidUploadIdMarker)); err != nil {
+			return err
+		}
+
+		// uploadId marker and key marker mismatch
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:         &bucket,
+			KeyMarker:      getPtr("obj-2"),
+			UploadIdMarker: uploads[4].UploadId,
+		})
+		cancel()
+		return checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidUploadIdMarker))
+	})
+}
+
+func ListMultipartUploads_keyMarker_not_from_list(s *S3Conf) error {
+	testName := "ListMultipartUploads_keyMarker_not_from_list"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		uploads := make([]types.MultipartUpload, 0, 9)
+		for _, mp := range []struct {
+			key   string
+			count int
+		}{
+			{"bar", 3},
+			{"baz", 4},
+			{"foo", 2},
+		} {
+			for range mp.count {
+				out, err := createMp(s3client, bucket, mp.key)
+				if err != nil {
+					return err
+				}
+				uploads = append(uploads, types.MultipartUpload{
+					Key:          out.Key,
+					UploadId:     out.UploadId,
+					StorageClass: types.StorageClassStandard,
+				})
+				if s.azureTests {
+					// add an artificial delay for azure tests
+					// as azure uploads all these mps with the same
+					// identical creation time
+					time.Sleep(time.Second)
+				}
+			}
+		}
+
+		// without uploadId marker
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:    &bucket,
+			KeyMarker: getPtr("bat"),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareMultipartUploads(uploads[3:], out.Uploads) {
+			return fmt.Errorf("expected the mp list to be %v, instead got %v", uploads[:3], out.Uploads)
+		}
+
+		// should start the listing after the specified uploadId marker
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err = s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:         &bucket,
+			KeyMarker:      getPtr("bat"),
+			UploadIdMarker: uploads[4].UploadId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareMultipartUploads(uploads[5:], out.Uploads) {
+			return fmt.Errorf("expected the mp list to be %v, instead got %v", uploads[5:], out.Uploads)
+		}
+
+		return nil
+	})
+}
+
+func ListMultipartUploads_delimiter_truncated(s *S3Conf) error {
+	testName := "ListMultipartUploads_delimiter_truncated"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		uploads := make([]types.MultipartUpload, 0, 6)
+		for _, key := range []string{
+			"abc/something",
+			"foo/bar/baz",
+			"foo/quxx",
+			"xyz/hello",
+			"zzz/bca",
+			"some/very/nested/mp/object",
+		} {
+			out, err := createMp(s3client, bucket, key)
+			if err != nil {
+				return err
+			}
+			uploads = append(uploads, types.MultipartUpload{
+				Key:          out.Key,
+				UploadId:     out.UploadId,
+				StorageClass: types.StorageClassStandard,
+			})
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:     &bucket,
+			Delimiter:  getPtr("/"),
+			MaxUploads: getPtr(int32(2)),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(out.Uploads) != 0 {
+			return fmt.Errorf("expected empty uplodas list, instead got %v", out.Uploads)
+		}
+		expectedCps := []string{"abc/", "foo/"}
+		if !comparePrefixes(expectedCps, out.CommonPrefixes) {
+			return fmt.Errorf("expected the common prefixes to be %v, instead got %v", expectedCps, out.CommonPrefixes)
+		}
+		if getString(out.NextKeyMarker) != "foo/" {
+			return fmt.Errorf("expected the next key marker to be 'foo/', instead got %s", getString(out.NextKeyMarker))
+		}
+		if getString(out.NextUploadIdMarker) != getString(uploads[1].UploadId) {
+			return fmt.Errorf("expected the next upload id marker to be %s, instead got %s", getString(uploads[1].UploadId), getString(out.NextUploadIdMarker))
+		}
+		if !*out.IsTruncated {
+			return fmt.Errorf("expected a truncated response")
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out2, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:         &bucket,
+			Delimiter:      getPtr("/"),
+			UploadIdMarker: out.NextUploadIdMarker,
+			KeyMarker:      out.NextKeyMarker,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(out2.Uploads) != 0 {
+			return fmt.Errorf("expected empty uplodas list, instead got %v", out2.Uploads)
+		}
+		expectedCps = []string{"foo/", "some/", "xyz/", "zzz/"}
+		if !comparePrefixes(expectedCps, out2.CommonPrefixes) {
+			return fmt.Errorf("expected the common prefixes to be %v, instead got %v", expectedCps, out2.CommonPrefixes)
+		}
+		if getString(out2.KeyMarker) != "foo/" {
+			return fmt.Errorf("expected key marker to be 'foo/', instead got %s", getString(out2.KeyMarker))
+		}
+		if getString(out2.UploadIdMarker) != getString(uploads[1].UploadId) {
+			return fmt.Errorf("expected the upload id marker to be %s, instead got %s", getString(uploads[1].UploadId), getString(out2.UploadIdMarker))
+		}
+		if getString(out2.NextKeyMarker) != "" {
+			return fmt.Errorf("expected empty next key marker, instead got %s", getString(out2.NextKeyMarker))
+		}
+		if getString(out2.NextUploadIdMarker) != "" {
+			return fmt.Errorf("expected empty next upload id marker, instead got %s", getString(out2.NextUploadIdMarker))
+		}
+		if *out2.IsTruncated {
+			return fmt.Errorf("expected a non-truncated response")
+		}
+
+		return nil
+	})
+}
+
+func ListMultipartUploads_prefix(s *S3Conf) error {
+	testName := "ListMultipartUploads_prefix"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		prefix := "foo"
+		uploads := make([]types.MultipartUpload, 0, 8)
+		for _, key := range []string{
+			"abc/something",
+			"foo/bar/baz",
+			"foo/quxx",
+			"hello/world",
+			"xyz/hello",
+			"zzz/bca",
+			"some/very/nested/mp/object",
+			"foo/xyz",
+		} {
+			out, err := createMp(s3client, bucket, key)
+			if err != nil {
+				return err
+			}
+
+			if strings.HasPrefix(key, prefix) {
+				uploads = append(uploads, types.MultipartUpload{
+					Key:          out.Key,
+					UploadId:     out.UploadId,
+					StorageClass: types.StorageClassStandard,
+				})
+			}
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket: &bucket,
+			Prefix: &prefix,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(out.Prefix) != prefix {
+			return fmt.Errorf("expected the prefix to be %s, instead got %s", prefix, getString(out.Prefix))
+		}
+		if !compareMultipartUploads(out.Uploads, uploads) {
+			return fmt.Errorf("expected the uploads list to be %v, instead got %v", uploads, out.Uploads)
+		}
+
+		return nil
+	})
+}
+
+func ListMultipartUploads_both_delimiter_and_prefix(s *S3Conf) error {
+	testName := "ListMultipartUploads_both_delimiter_and_prefix"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		for _, key := range []string{
+			"foo/abc/bbb/aaa/c",
+			"abc/something",
+			"foo/bar/baz",
+			"foo/quxx",
+			"hello/world",
+			"foo/random/object",
+			"foo/random/another/object",
+			"xyz/hello",
+			"zzz/bca",
+			"some/very/nested/mp/object",
+			"foo/xyz",
+		} {
+			_, err := createMp(s3client, bucket, key)
+			if err != nil {
+				return err
+			}
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:    &bucket,
+			Delimiter: getPtr("/"),
+			Prefix:    getPtr("foo/"),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		expectedCps := []string{"foo/abc/", "foo/bar/", "foo/random/"}
+		if !comparePrefixes(expectedCps, out.CommonPrefixes) {
+			return fmt.Errorf("expected the common prefixes to be %v, instead got %v", expectedCps, out.CommonPrefixes)
+		}
+
+		return nil
+	})
+}
+
+func ListMultipartUploads_delimiter_no_matches(s *S3Conf) error {
+	testName := "ListMultipartUploads_delimiter_no_matches"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		uploads := make([]types.MultipartUpload, 0, 8)
+		for _, key := range []string{
+			"abc/something",
+			"foo/bar/baz",
+			"foo/quxx",
+			"hello/world",
+			"xyz/hello",
+			"zzz/bca",
+			"some/very/nested/mp/object",
+			"foo/xyz",
+		} {
+			out, err := createMp(s3client, bucket, key)
+			if err != nil {
+				return err
+			}
+
+			uploads = append(uploads, types.MultipartUpload{
+				Key:          out.Key,
+				UploadId:     out.UploadId,
+				StorageClass: types.StorageClassStandard,
+			})
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+			Bucket:    &bucket,
+			Delimiter: getPtr("delim"),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		sort.SliceStable(uploads, func(i, j int) bool {
+			return *uploads[i].Key < *uploads[j].Key
+		})
+
+		if !compareMultipartUploads(uploads, out.Uploads) {
+			return fmt.Errorf("expected the uploads to be %v, instead got %v", uploads, out.Uploads)
+		}
+		if len(out.CommonPrefixes) != 0 {
+			return fmt.Errorf("expected empty common prefixes, instead got %v", out.CommonPrefixes)
+		}
+
 		return nil
 	})
 }
@@ -302,52 +645,3 @@ func ListMultipartUploads_with_checksums(s *S3Conf) error {
 		return nil
 	})
 }
-
-func ListMultipartUploads_success(s *S3Conf) error {
-	testName := "ListMultipartUploads_success"
-	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		obj1, obj2 := "my-obj-1", "my-obj-2"
-		out1, err := createMp(s3client, bucket, obj1)
-		if err != nil {
-			return err
-		}
-
-		out2, err := createMp(s3client, bucket, obj2)
-		if err != nil {
-			return err
-		}
-
-		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-		out, err := s3client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
-			Bucket: &bucket,
-		})
-		cancel()
-		if err != nil {
-			return err
-		}
-
-		expected := []types.MultipartUpload{
-			{
-				Key:          &obj1,
-				UploadId:     out1.UploadId,
-				StorageClass: types.StorageClassStandard,
-			},
-			{
-				Key:          &obj2,
-				UploadId:     out2.UploadId,
-				StorageClass: types.StorageClassStandard,
-			},
-		}
-
-		if len(out.Uploads) != 2 {
-			return fmt.Errorf("expected 2 upload, instead got %v",
-				len(out.Uploads))
-		}
-		if ok := compareMultipartUploads(out.Uploads, expected); !ok {
-			return fmt.Errorf("expected uploads %v, instead got %v",
-				expected, out.Uploads)
-		}
-
-		return nil
-	})
-}
diff --git a/tests/integration/group-tests.go b/tests/integration/group-tests.go
index b7d5db5..5bffa33 100644
--- a/tests/integration/group-tests.go
+++ b/tests/integration/group-tests.go
@@ -459,13 +459,17 @@ func TestListMultipartUploads(ts *TestState) {
 	ts.Run(ListMultipartUploads_invalid_max_uploads)
 	ts.Run(ListMultipartUploads_max_uploads)
 	ts.Run(ListMultipartUploads_exceeding_max_uploads)
-	ts.Run(ListMultipartUploads_incorrect_next_key_marker)
 	ts.Run(ListMultipartUploads_ignore_upload_id_marker)
+	ts.Run(ListMultipartUploads_invalid_uploadId_marker)
+	ts.Run(ListMultipartUploads_keyMarker_not_from_list)
+	ts.Run(ListMultipartUploads_delimiter_truncated)
+	ts.Run(ListMultipartUploads_prefix)
+	ts.Run(ListMultipartUploads_both_delimiter_and_prefix)
+	ts.Run(ListMultipartUploads_delimiter_no_matches)
 	//TODO: remove the condition after implementing checksums in azure
 	if !ts.conf.azureTests {
 		ts.Run(ListMultipartUploads_with_checksums)
 	}
-	ts.Run(ListMultipartUploads_success)
 }
 
 func TestAbortMultipartUpload(ts *TestState) {
@@ -1473,10 +1477,13 @@ func GetIntTests() IntTests {
 		"ListMultipartUploads_invalid_max_uploads":                                 ListMultipartUploads_invalid_max_uploads,
 		"ListMultipartUploads_max_uploads":                                         ListMultipartUploads_max_uploads,
 		"ListMultipartUploads_exceeding_max_uploads":                               ListMultipartUploads_exceeding_max_uploads,
-		"ListMultipartUploads_incorrect_next_key_marker":                           ListMultipartUploads_incorrect_next_key_marker,
 		"ListMultipartUploads_ignore_upload_id_marker":                             ListMultipartUploads_ignore_upload_id_marker,
+		"ListMultipartUploads_invalid_uploadId_marker":                             ListMultipartUploads_invalid_uploadId_marker,
+		"ListMultipartUploads_keyMarker_not_from_list":                             ListMultipartUploads_keyMarker_not_from_list,
+		"ListMultipartUploads_delimiter_truncated":                                 ListMultipartUploads_delimiter_truncated,
+		"ListMultipartUploads_prefix":                                              ListMultipartUploads_prefix,
+		"ListMultipartUploads_both_delimiter_and_prefix":                           ListMultipartUploads_both_delimiter_and_prefix,
 		"ListMultipartUploads_with_checksums":                                      ListMultipartUploads_with_checksums,
-		"ListMultipartUploads_success":                                             ListMultipartUploads_success,
 		"AbortMultipartUpload_non_existing_bucket":                                 AbortMultipartUpload_non_existing_bucket,
 		"AbortMultipartUpload_incorrect_uploadId":                                  AbortMultipartUpload_incorrect_uploadId,
 		"AbortMultipartUpload_incorrect_object_key":                                AbortMultipartUpload_incorrect_object_key,
diff --git a/tests/integration/utils.go b/tests/integration/utils.go
index db06121..c9d69cd 100644
--- a/tests/integration/utils.go
+++ b/tests/integration/utils.go
@@ -838,14 +838,13 @@ func comparePrefixes(list1 []string, list2 []types.CommonPrefix) bool {
 		return false
 	}
 
-	elementMap := make(map[string]bool)
-
-	for _, elem := range list1 {
-		elementMap[elem] = true
-	}
-
-	for _, elem := range list2 {
-		if _, found := elementMap[*elem.Prefix]; !found {
+	for i, prefix := range list1 {
+		if list2[i].Prefix == nil {
+			fmt.Printf("unexpected nil prefix on index %v", i)
+			return false
+		}
+		if *list2[i].Prefix != prefix {
+			fmt.Printf("prefix mismatch on index %v: expected %s, got %v", i, prefix, *list2[i].Prefix)
 			return false
 		}
 	}