From 6759cdf38dbec3fbe951e328e5977f2b2cda4ecb Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Fri, 10 Apr 2026 00:15:39 +0530 Subject: [PATCH 01/11] HDDS-14665. Add upgrade handling to multipart requests --- .../s3/multipart/S3MultipartUploadAbortRequest.java | 8 ++++++++ .../s3/multipart/S3MultipartUploadCommitPartRequest.java | 8 ++++++++ .../s3/multipart/S3MultipartUploadCompleteRequest.java | 7 +++++++ .../apache/hadoop/ozone/om/upgrade/OMLayoutFeature.java | 4 +++- 4 files changed, 26 insertions(+), 1 deletion(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java index a9aeff0ac5d1..e2e910629147 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java @@ -165,6 +165,14 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } + if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + && multipartKeyInfo.getSchemaVersion() != 0) { + throw new OMException("MPU parts-table split behavior is not allowed " + + "before cluster finalization.", + OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); + } + + multipartKeyInfo = multipartKeyInfo.toBuilder() .setUpdateID(trxnLogIndex) .build(); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java index ac123ff680ac..799f82dc3b6b 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java @@ -147,6 +147,14 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut multipartKeyInfo = omMetadataManager.getMultipartInfoTable() .get(multipartKey); + + if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + && multipartKeyInfo.getSchemaVersion() != 0) { + throw new OMException("MPU parts-table split behavior is not allowed " + + "before cluster finalization for commit part request.", + OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); + } + openKey = getOpenKey(volumeName, bucketName, keyName, omMetadataManager, clientID); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java index cadd7f80b62f..25e105a496a2 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java @@ -251,6 +251,13 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OmMultipartKeyInfo multipartKeyInfo = omMetadataManager .getMultipartInfoTable().get(multipartKey); + + if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + && multipartKeyInfo.getSchemaVersion() != 0) { + throw new OMException("MPU parts-table split behavior is not allowed " + + "before cluster finalization for commit part request.", + OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); + } String ozoneKey = omMetadataManager.getOzoneKey( volumeName, bucketName, keyName); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/upgrade/OMLayoutFeature.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/upgrade/OMLayoutFeature.java index ef99b453b7f0..c693a719e138 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/upgrade/OMLayoutFeature.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/upgrade/OMLayoutFeature.java @@ -44,7 +44,9 @@ public enum OMLayoutFeature implements LayoutFeature { QUOTA(6, "Ozone quota re-calculate"), HBASE_SUPPORT(7, "Full support of hsync, lease recovery and listOpenFiles APIs for HBase"), DELEGATION_TOKEN_SYMMETRIC_SIGN(8, "Delegation token signed by symmetric key"), - SNAPSHOT_DEFRAG(9, "Supporting defragmentation of snapshot"); + SNAPSHOT_DEFRAG(9, "Supporting defragmentation of snapshot"), + + MPU_PARTS_TABLE_SPLIT(10, "Split multipart table into separate table for parts and key"); /////////////////////////////// ///////////////////////////// From f031d9014682ee51169ba9295d0e3d4460ef3364 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Fri, 10 Apr 2026 00:29:11 +0530 Subject: [PATCH 02/11] Add tests for upgrade handling --- .../s3/multipart/TestS3MultipartRequest.java | 44 +++++++++++++++++++ .../TestS3MultipartUploadAbortRequest.java | 32 ++++++++++++++ ...estS3MultipartUploadCommitPartRequest.java | 36 +++++++++++++++ .../TestS3MultipartUploadCompleteRequest.java | 30 +++++++++++++ 4 files changed, 142 insertions(+) diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartRequest.java index 38749f6812a6..71167e37a9d9 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartRequest.java @@ -34,6 +34,8 @@ import java.util.Map; import org.apache.hadoop.hdds.client.ReplicationConfig; import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.utils.db.cache.CacheKey; +import org.apache.hadoop.hdds.utils.db.cache.CacheValue; import org.apache.hadoop.ozone.audit.AuditLogger; import org.apache.hadoop.ozone.audit.AuditMessage; import org.apache.hadoop.ozone.om.IOmMetadataReader; @@ -47,8 +49,10 @@ import org.apache.hadoop.ozone.om.ResolvedBucket; import org.apache.hadoop.ozone.om.helpers.BucketLayout; import org.apache.hadoop.ozone.om.helpers.KeyValueUtil; +import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo; import org.apache.hadoop.ozone.om.request.OMClientRequest; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; +import org.apache.hadoop.ozone.om.response.OMClientResponse; import org.apache.hadoop.ozone.om.upgrade.OMLayoutVersionManager; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocation; @@ -302,6 +306,46 @@ protected OMRequest doPreExecuteCompleteMPU( } + /** + * Initiate an MPU and optionally rewrite the stored multipart metadata to a + * specific schema version. + * + *

The schema version rewrite lets tests emulate post-finalization MPU + * entries without needing the rest of the upgrade pipeline.

+ */ + protected String initiateMultipartUploadWithSchemaVersion( + String volumeName, String bucketName, String keyName, + byte schemaVersion) throws Exception { + OMRequest initiateMPURequest = + doPreExecuteInitiateMPU(volumeName, bucketName, keyName); + + S3InitiateMultipartUploadRequest s3InitiateMultipartUploadRequest = + getS3InitiateMultipartUploadReq(initiateMPURequest); + + OMClientResponse omClientResponse = + s3InitiateMultipartUploadRequest.validateAndUpdateCache(ozoneManager, + 1L); + + String multipartUploadID = omClientResponse.getOMResponse() + .getInitiateMultiPartUploadResponse().getMultipartUploadID(); + + if (schemaVersion != 0) { + String multipartKey = omMetadataManager.getMultipartKey(volumeName, + bucketName, keyName, multipartUploadID); + OmMultipartKeyInfo multipartKeyInfo = omMetadataManager + .getMultipartInfoTable().get(multipartKey); + assertNotNull(multipartKeyInfo); + + omMetadataManager.getMultipartInfoTable().addCacheEntry( + new CacheKey<>(multipartKey), + CacheValue.get(2L, multipartKeyInfo.toBuilder() + .setSchemaVersion(schemaVersion) + .build())); + } + + return multipartUploadID; + } + /** * Perform preExecute of Initiate Multipart upload request for given * volume, bucket and key name. diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java index d9f84523204e..4e3a6a4f4998 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java @@ -95,6 +95,38 @@ public void testValidateAndUpdateCache() throws Exception { } + @Test + public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization() + throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = getKeyName(); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + createParentPath(volumeName, bucketName); + + String multipartUploadID = + initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, + keyName, (byte) 1); + + OMRequest abortMPURequest = + doPreExecuteAbortMPU(volumeName, bucketName, keyName, + multipartUploadID); + + S3MultipartUploadAbortRequest s3MultipartUploadAbortRequest = + getS3MultipartUploadAbortReq(abortMPURequest); + + // The multipart metadata exists, but schema version 1 is not allowed yet. + OMClientResponse omClientResponse = + s3MultipartUploadAbortRequest.validateAndUpdateCache(ozoneManager, 2L); + + assertEquals(OzoneManagerProtocolProtos.Status + .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, + omClientResponse.getOMResponse().getStatus()); + } + @Test public void testValidateAndUpdateCacheMultipartNotFound() throws Exception { String volumeName = UUID.randomUUID().toString(); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java index b5d78662507a..808bfa22a62c 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java @@ -129,6 +129,42 @@ public void testValidateAndUpdateCacheSuccess() throws Exception { .get(partKey)); } + @Test + public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization() + throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = getKeyName(); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + createParentPath(volumeName, bucketName); + + String multipartUploadID = + initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, + keyName, (byte) 1); + + long clientID = Time.now(); + OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName, + bucketName, keyName, clientID, multipartUploadID, 1); + + S3MultipartUploadCommitPartRequest s3MultipartUploadCommitPartRequest = + getS3MultipartUploadCommitReq(commitMultipartRequest); + + addKeyToOpenKeyTable(volumeName, bucketName, keyName, clientID); + + // Regular part metadata is present; the upgrade gate should still reject + // this schema version before commit proceeds. + OMClientResponse omClientResponse = + s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, + 2L); + + assertEquals(OzoneManagerProtocolProtos.Status + .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, + omClientResponse.getOMResponse().getStatus()); + } + @Test public void testValidateAndUpdateCacheMultipartNotFound() throws Exception { String volumeName = UUID.randomUUID().toString(); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java index dbf82cc18ea0..940e310eed9d 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java @@ -206,6 +206,36 @@ private String checkValidateAndUpdateCacheSuccess(String volumeName, return multipartUploadID; } + @Test + public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization() + throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = getKeyName(); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + String multipartUploadID = + initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, + keyName, (byte) 1); + + // The request is still rejected before the empty part-list validation. + OMRequest completeMultipartRequest = doPreExecuteCompleteMPU(volumeName, + bucketName, keyName, multipartUploadID, new ArrayList<>()); + + S3MultipartUploadCompleteRequest s3MultipartUploadCompleteRequest = + getS3MultipartUploadCompleteReq(completeMultipartRequest); + + OMClientResponse omClientResponse = + s3MultipartUploadCompleteRequest.validateAndUpdateCache(ozoneManager, + 3L); + + assertEquals(OzoneManagerProtocolProtos.Status + .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, + omClientResponse.getOMResponse().getStatus()); + } + protected void addVolumeAndBucket(String volumeName, String bucketName) throws Exception { OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, From 598bdf46fba1bac744aae075c65b749843aeacaa Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Tue, 28 Apr 2026 14:10:58 +0530 Subject: [PATCH 03/11] Fix checkstyle issue --- .../s3/multipart/S3MultipartUploadAbortRequest.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java index e2e910629147..5256f5cf2cca 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java @@ -165,12 +165,12 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) - && multipartKeyInfo.getSchemaVersion() != 0) { - throw new OMException("MPU parts-table split behavior is not allowed " + - "before cluster finalization.", - OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); - } + if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + && multipartKeyInfo.getSchemaVersion() != 0) { + throw new OMException("MPU parts-table split behavior is not allowed " + + "before cluster finalization.", + OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); + } multipartKeyInfo = multipartKeyInfo.toBuilder() From 9cc3a55b2646ab04f3c15be818d8deccd5cc2c4d Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Thu, 14 May 2026 12:23:37 +0530 Subject: [PATCH 04/11] Fix null check bug --- .../S3MultipartUploadCommitPartRequest.java | 27 +++++++++---------- .../S3MultipartUploadCompleteRequest.java | 14 +++++----- 2 files changed, 20 insertions(+), 21 deletions(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java index 799f82dc3b6b..5c0c44900810 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java @@ -147,14 +147,25 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut multipartKeyInfo = omMetadataManager.getMultipartInfoTable() .get(multipartKey); - + + if (multipartKeyInfo == null) { + // This can occur when user started uploading part by the time commit + // of that part happens, in between the user might have requested + // abort multipart upload. If we just throw exception, then the data + // will not be garbage collected, so move this part to delete table + // and throw error + // Move this part to delete table. + throw new OMException("No such Multipart upload is with specified " + + "uploadId " + uploadID, + OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); + } + if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) && multipartKeyInfo.getSchemaVersion() != 0) { throw new OMException("MPU parts-table split behavior is not allowed " + "before cluster finalization for commit part request.", OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); } - openKey = getOpenKey(volumeName, bucketName, keyName, omMetadataManager, clientID); @@ -189,18 +200,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut int partNumber = keyArgs.getMultipartNumber(); partName = getPartName(ozoneKey, uploadID, partNumber); - if (multipartKeyInfo == null) { - // This can occur when user started uploading part by the time commit - // of that part happens, in between the user might have requested - // abort multipart upload. If we just throw exception, then the data - // will not be garbage collected, so move this part to delete table - // and throw error - // Move this part to delete table. - throw new OMException("No such Multipart upload is with specified " + - "uploadId " + uploadID, - OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); - } - oldPartKeyInfo = multipartKeyInfo.getPartKeyInfo(partNumber); // Build this multipart upload part info. diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java index 25e105a496a2..70dd28ca5a9b 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java @@ -251,13 +251,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OmMultipartKeyInfo multipartKeyInfo = omMetadataManager .getMultipartInfoTable().get(multipartKey); - - if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) - && multipartKeyInfo.getSchemaVersion() != 0) { - throw new OMException("MPU parts-table split behavior is not allowed " + - "before cluster finalization for commit part request.", - OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); - } String ozoneKey = omMetadataManager.getOzoneKey( volumeName, bucketName, keyName); @@ -276,6 +269,13 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } + if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + && multipartKeyInfo.getSchemaVersion() != 0) { + throw new OMException("MPU parts-table split behavior is not allowed " + + "before cluster finalization for commit part request.", + OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); + } + if (!partsList.isEmpty()) { final OmMultipartKeyInfo.PartKeyInfoMap partKeyInfoMap = multipartKeyInfo.getPartKeyInfoMap(); From 1cafdd40e1808ddd34cb4bc24e372a2d4b94bc61 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Thu, 14 May 2026 15:25:57 +0530 Subject: [PATCH 05/11] Fix MPU test --- ...estS3MultipartUploadCommitPartRequest.java | 22 +++++++++++-------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java index 808bfa22a62c..eb405fb536ac 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java @@ -33,7 +33,6 @@ import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; -import org.apache.hadoop.ozone.om.helpers.BucketLayout; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo; @@ -210,9 +209,19 @@ public void testValidateAndUpdateCacheKeyNotFound() throws Exception { OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout()); + createParentPath(volumeName, bucketName); + + OMRequest initiateMPURequest = doPreExecuteInitiateMPU(volumeName, + bucketName, keyName); + S3InitiateMultipartUploadRequest s3InitiateMultipartUploadRequest = + getS3InitiateMultipartUploadReq(initiateMPURequest); + OMClientResponse initiateResponse = + s3InitiateMultipartUploadRequest.validateAndUpdateCache(ozoneManager, + 1L); long clientID = Time.now(); - String multipartUploadID = UUID.randomUUID().toString(); + String multipartUploadID = initiateResponse.getOMResponse() + .getInitiateMultiPartUploadResponse().getMultipartUploadID(); OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName, bucketName, keyName, clientID, multipartUploadID, 1); @@ -227,13 +236,8 @@ public void testValidateAndUpdateCacheKeyNotFound() throws Exception { OMClientResponse omClientResponse = s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, 2L); - if (getBucketLayout() == BucketLayout.FILE_SYSTEM_OPTIMIZED) { - assertSame(omClientResponse.getOMResponse().getStatus(), - OzoneManagerProtocolProtos.Status.DIRECTORY_NOT_FOUND); - } else { - assertSame(omClientResponse.getOMResponse().getStatus(), - OzoneManagerProtocolProtos.Status.KEY_NOT_FOUND); - } + assertSame(omClientResponse.getOMResponse().getStatus(), + OzoneManagerProtocolProtos.Status.KEY_NOT_FOUND); } From 631050666dc77ff052261af388299d88fea338e6 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Fri, 5 Jun 2026 13:21:07 +0530 Subject: [PATCH 06/11] Address Review comments --- .../ozone/om/helpers/OmMultipartKeyInfo.java | 30 +++++++--- .../om/helpers/TestOmMultipartKeyInfo.java | 16 ++++++ .../S3MultipartUploadAbortRequest.java | 7 ++- .../S3MultipartUploadCommitPartRequest.java | 44 +++++++++------ .../S3MultipartUploadCompleteRequest.java | 12 ++-- .../TestS3MultipartUploadAbortRequest.java | 39 +++++++++++++ ...estS3MultipartUploadCommitPartRequest.java | 55 +++++++++++++++++++ ...ltipartUploadCommitPartRequestWithFSO.java | 30 ++++++++++ .../TestS3MultipartUploadCompleteRequest.java | 37 +++++++++++++ 9 files changed, 239 insertions(+), 31 deletions(-) diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java index 88036d3e66a6..8802d4eb0c00 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java @@ -87,6 +87,8 @@ public final class OmMultipartKeyInfo extends WithObjectID implements CopyObject // This stores the schema version of the multipart key. // 0 - Legacy Schema -> Uses the same table to store the multipart part info // 1 - New Schema -> Uses a separate table to store the multipart part info + private static final int LEGACY_SCHEMA_VERSION = 0; + private static final int SPLIT_PARTS_SCHEMA_VERSION = 1; private final byte schemaVersion; public static Codec getCodec() { @@ -258,7 +260,7 @@ public PartKeyInfoMap getPartKeyInfoMap() { } public void addPartKeyInfo(PartKeyInfo partKeyInfo) { - if (schemaVersion == 1) { + if (schemaVersion == SPLIT_PARTS_SCHEMA_VERSION) { throw new IllegalStateException( "PartKeyInfoMap is not supported for schemaVersion 1"); } @@ -314,7 +316,7 @@ public Builder(OmMultipartKeyInfo multipartKeyInfo) { this.acls = AclListBuilder.of(multipartKeyInfo.acls); this.partKeyInfoList = new TreeMap<>(); - if (multipartKeyInfo.getSchemaVersion() == 0) { + if (multipartKeyInfo.getSchemaVersion() == LEGACY_SCHEMA_VERSION) { for (PartKeyInfo partKeyInfo : multipartKeyInfo.partKeyInfoMap) { this.partKeyInfoList.put(partKeyInfo.getPartNumber(), partKeyInfo); } @@ -409,7 +411,8 @@ public Builder setParentID(long parentObjId) { } public Builder setSchemaVersion(byte schemaVersion) { - this.schemaVersion = schemaVersion; + this.schemaVersion = + validateAndConvertSchemaVersion(Byte.toUnsignedInt(schemaVersion)); return this; } @@ -427,7 +430,8 @@ protected OmMultipartKeyInfo buildObject() { public static Builder builderFromProto( MultipartKeyInfo multipartKeyInfo) { final SortedMap list = new TreeMap<>(); - if (!multipartKeyInfo.hasSchemaVersion() || multipartKeyInfo.getSchemaVersion() == 0) { + if (!multipartKeyInfo.hasSchemaVersion() + || multipartKeyInfo.getSchemaVersion() == LEGACY_SCHEMA_VERSION) { multipartKeyInfo.getPartKeyInfoListList().forEach(partKeyInfo -> list.put(partKeyInfo.getPartNumber(), partKeyInfo)); } @@ -455,7 +459,8 @@ public static Builder builderFromProto( .setObjectID(multipartKeyInfo.getObjectID()) .setUpdateID(multipartKeyInfo.getUpdateID()) .setParentID(multipartKeyInfo.getParentID()) - .setSchemaVersion((byte) multipartKeyInfo.getSchemaVersion()); + .setSchemaVersion(validateAndConvertSchemaVersion( + multipartKeyInfo.getSchemaVersion())); } /** @@ -473,7 +478,9 @@ public static OmMultipartKeyInfo getFromProto( * @return MultipartKeyInfo */ public MultipartKeyInfo getProto() { - if (schemaVersion == 1 && partKeyInfoMap != null && partKeyInfoMap.size() > 0) { + if (schemaVersion == SPLIT_PARTS_SCHEMA_VERSION + && partKeyInfoMap != null + && partKeyInfoMap.size() > 0) { throw new IllegalStateException( "PartKeyInfoMap must be empty for schemaVersion 1"); } @@ -507,12 +514,21 @@ public MultipartKeyInfo getProto() { } builder.addAllAcls(OzoneAclUtil.toProtobuf(acls)); - if (schemaVersion == 0) { + if (schemaVersion == LEGACY_SCHEMA_VERSION) { builder.addAllPartKeyInfoList(partKeyInfoMap); } return builder.build(); } + private static byte validateAndConvertSchemaVersion(int schemaVersion) { + if (schemaVersion != LEGACY_SCHEMA_VERSION + && schemaVersion != SPLIT_PARTS_SCHEMA_VERSION) { + throw new IllegalArgumentException("Unsupported schemaVersion: " + + schemaVersion + ". Expected one of [0, 1]."); + } + return (byte) schemaVersion; + } + @Override public String getObjectInfo() { return getProto().toString(); diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java index 3a5658dd5bde..ee8d7de50a73 100644 --- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java +++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java @@ -145,6 +145,22 @@ public void getProtoRejectsLegacyPartListForSchemaVersionOne() { assertThrows(IllegalStateException.class, subject::getProto); } + @Test + public void builderFromProtoRejectsUnsupportedSchemaVersion() { + OmMultipartKeyInfo subject = createSubject() + .setReplicationConfig(StandaloneReplicationConfig.getInstance( + HddsProtos.ReplicationFactor.ONE)) + .build(); + + OzoneManagerProtocolProtos.MultipartKeyInfo invalidProto = subject.getProto() + .toBuilder() + .setSchemaVersion(256) + .build(); + + assertThrows(IllegalArgumentException.class, + () -> OmMultipartKeyInfo.getFromProto(invalidProto)); + } + private static OmMultipartKeyInfo.Builder createSubject() { return new OmMultipartKeyInfo.Builder() .setUploadID(UUID.randomUUID().toString()) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java index 5256f5cf2cca..d8ecfd014f5d 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java @@ -165,14 +165,17 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + // This gate runs in the replicated apply path, so the check must remain + // deterministic across replicas for a given log index. MLV advances only + // via the Ratis-logged finalize-upgrade request. + if (!ozoneManager.getVersionManager() + .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) && multipartKeyInfo.getSchemaVersion() != 0) { throw new OMException("MPU parts-table split behavior is not allowed " + "before cluster finalization.", OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); } - multipartKeyInfo = multipartKeyInfo.toBuilder() .setUpdateID(trxnLogIndex) .build(); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java index 5c0c44900810..2f52c0e5c16b 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java @@ -148,37 +148,45 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut multipartKeyInfo = omMetadataManager.getMultipartInfoTable() .get(multipartKey); + openKey = getOpenKey(volumeName, bucketName, keyName, omMetadataManager, + clientID); + + String ozoneKey = omMetadataManager.getOzoneKey( + volumeName, bucketName, keyName); + + omKeyInfo = getOmKeyInfo(omMetadataManager, openKey, keyName); + + if (omKeyInfo == null) { + throw new OMException("Failed to commit Multipart Upload key, as " + + openKey + "entry is not found in the openKey table", + KEY_NOT_FOUND); + } + if (multipartKeyInfo == null) { // This can occur when user started uploading part by the time commit // of that part happens, in between the user might have requested // abort multipart upload. If we just throw exception, then the data // will not be garbage collected, so move this part to delete table - // and throw error - // Move this part to delete table. + // and throw error. throw new OMException("No such Multipart upload is with specified " + "uploadId " + uploadID, OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + // This gate runs in the replicated apply path, so the check must remain + // deterministic across replicas for a given log index. MLV advances only + // via the Ratis-logged finalize-upgrade request. + if (!ozoneManager.getVersionManager() + .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) && multipartKeyInfo.getSchemaVersion() != 0) { throw new OMException("MPU parts-table split behavior is not allowed " + - "before cluster finalization for commit part request.", - OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); + "before cluster finalization for commit part request.", + OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); } - - openKey = getOpenKey(volumeName, bucketName, keyName, omMetadataManager, - clientID); - - String ozoneKey = omMetadataManager.getOzoneKey( - volumeName, bucketName, keyName); - - omKeyInfo = getOmKeyInfo(omMetadataManager, openKey, keyName); - - if (omKeyInfo == null) { - throw new OMException("Failed to commit Multipart Upload key, as " + - openKey + "entry is not found in the openKey table", - KEY_NOT_FOUND); + if (multipartKeyInfo.getSchemaVersion() == 1) { + throw new OMException("MPU parts-table split commit path is not " + + "supported in this write flow.", + OMException.ResultCodes.NOT_SUPPORTED_OPERATION); } // Add/Update user defined metadata. // Set the UpdateID to current transactionLogIndex diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java index 5811a66ede9b..5f2708b0f3f1 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java @@ -269,18 +269,22 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - if (!ozoneManager.getVersionManager().isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + // This gate runs in the replicated apply path, so the check must remain + // deterministic across replicas for a given log index. MLV advances only + // via the Ratis-logged finalize-upgrade request. + if (!ozoneManager.getVersionManager() + .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) && multipartKeyInfo.getSchemaVersion() != 0) { throw new OMException("MPU parts-table split behavior is not allowed " + - "before cluster finalization for commit part request.", - OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); + "before cluster finalization for complete request.", + OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); } // Conditional write validation (If-None-Match / If-Match). // BUCKET_LOCK is held, so validation and commit are atomic. // Only 412 PreconditionFailed is possible; 409 Conflict cannot occur. OmKeyInfo existingKeyInfo = - getOmKeyInfoFromKeyTable(dbOzoneKey, keyName, omMetadataManager); + omMetadataManager.getKeyTable(getBucketLayout()).get(dbOzoneKey); validateAtomicRewrite(existingKeyInfo, keyArgs); validateIfMatchETag(keyArgs, existingKeyInfo); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java index 4e3a6a4f4998..aca99c6847ff 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java @@ -18,8 +18,10 @@ package org.apache.hadoop.ozone.om.request.s3.multipart; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.when; import java.io.IOException; import java.util.UUID; @@ -27,6 +29,7 @@ import org.apache.hadoop.hdds.utils.db.cache.CacheValue; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; import org.apache.hadoop.ozone.om.response.OMClientResponse; +import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; import org.junit.jupiter.api.Test; @@ -127,6 +130,42 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( omClientResponse.getOMResponse().getStatus()); } + @Test + public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() + throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = getKeyName(); + + when(ozoneManager.getVersionManager() + .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT)).thenReturn(true); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + createParentPath(volumeName, bucketName); + + String multipartUploadID = + initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, + keyName, (byte) 1); + + OMRequest abortMPURequest = + doPreExecuteAbortMPU(volumeName, bucketName, keyName, + multipartUploadID); + + S3MultipartUploadAbortRequest s3MultipartUploadAbortRequest = + getS3MultipartUploadAbortReq(abortMPURequest); + + OMClientResponse omClientResponse = + s3MultipartUploadAbortRequest.validateAndUpdateCache(ozoneManager, 2L); + + assertEquals(OzoneManagerProtocolProtos.Status.OK, + omClientResponse.getOMResponse().getStatus()); + assertNotEquals(OzoneManagerProtocolProtos.Status + .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, + omClientResponse.getOMResponse().getStatus()); + } + @Test public void testValidateAndUpdateCacheMultipartNotFound() throws Exception { String volumeName = UUID.randomUUID().toString(); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java index eb405fb536ac..5d4ef02c1a95 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java @@ -23,6 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.when; import java.io.IOException; import java.util.ArrayList; @@ -33,6 +34,7 @@ import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; +import org.apache.hadoop.hdds.utils.db.BatchOperation; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo; @@ -40,6 +42,7 @@ import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; import org.apache.hadoop.ozone.om.response.OMClientResponse; import org.apache.hadoop.ozone.om.response.s3.multipart.S3MultipartUploadCommitPartResponse; +import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocation; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; @@ -164,6 +167,45 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( omClientResponse.getOMResponse().getStatus()); } + @Test + public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() + throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = getKeyName(); + + when(ozoneManager.getVersionManager() + .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT)).thenReturn(true); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + createParentPath(volumeName, bucketName); + + String multipartUploadID = + initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, + keyName, (byte) 1); + + long clientID = Time.now(); + OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName, + bucketName, keyName, clientID, multipartUploadID, 1); + + S3MultipartUploadCommitPartRequest s3MultipartUploadCommitPartRequest = + getS3MultipartUploadCommitReq(commitMultipartRequest); + + addKeyToOpenKeyTable(volumeName, bucketName, keyName, clientID); + + OMClientResponse omClientResponse = + s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, + 2L); + + assertNotEquals(OzoneManagerProtocolProtos.Status + .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, + omClientResponse.getOMResponse().getStatus()); + assertEquals(OzoneManagerProtocolProtos.Status.NOT_SUPPORTED_OPERATION, + omClientResponse.getOMResponse().getStatus()); + } + @Test public void testValidateAndUpdateCacheMultipartNotFound() throws Exception { String volumeName = UUID.randomUUID().toString(); @@ -186,6 +228,10 @@ public void testValidateAndUpdateCacheMultipartNotFound() throws Exception { // Add key to open key table. addKeyToOpenKeyTable(volumeName, bucketName, keyName, clientID); + String openKey = getOpenKey(volumeName, bucketName, keyName, clientID); + OmKeyInfo openPartKeyInfo = omMetadataManager.getOpenKeyTable( + s3MultipartUploadCommitPartRequest.getBucketLayout()).get(openKey); + assertNotNull(openPartKeyInfo); OMClientResponse omClientResponse = s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, 2L); @@ -198,6 +244,15 @@ public void testValidateAndUpdateCacheMultipartNotFound() throws Exception { assertNull(omMetadataManager.getMultipartInfoTable().get(multipartKey)); + BatchOperation batchOperation = omMetadataManager.getStore() + .initBatchOperation(); + omClientResponse.checkAndUpdateDB(omMetadataManager, batchOperation); + omMetadataManager.getStore().commitBatchOperation(batchOperation); + + String deleteKey = omMetadataManager.getOzoneDeletePathKey( + openPartKeyInfo.getObjectID(), multipartKey); + assertNotNull(omMetadataManager.getDeletedTable().get(deleteKey)); + } @Test diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequestWithFSO.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequestWithFSO.java index 4bff04d89138..6c823dcdaccb 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequestWithFSO.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequestWithFSO.java @@ -18,6 +18,7 @@ package org.apache.hadoop.ozone.om.request.s3.multipart; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -35,8 +36,12 @@ import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup; import org.apache.hadoop.ozone.om.helpers.OzoneFSUtils; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; +import org.apache.hadoop.ozone.om.response.OMClientResponse; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.util.Time; +import org.junit.jupiter.api.Test; /** * Tests S3 Multipart upload commit part request. @@ -71,6 +76,31 @@ protected String getKeyName() { return dirName + UUID.randomUUID().toString(); } + @Test + public void testValidateAndUpdateCacheDirectoryNotFound() throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = getKeyName(); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + long clientID = Time.now(); + String multipartUploadID = UUID.randomUUID().toString(); + OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName, + bucketName, keyName, clientID, multipartUploadID, 1); + + S3MultipartUploadCommitPartRequest s3MultipartUploadCommitPartRequest = + getS3MultipartUploadCommitReq(commitMultipartRequest); + + OMClientResponse omClientResponse = + s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, + 2L); + + assertEquals(OzoneManagerProtocolProtos.Status.DIRECTORY_NOT_FOUND, + omClientResponse.getOMResponse().getStatus()); + } + @Override protected void addKeyToOpenKeyTable(String volumeName, String bucketName, String keyName, long clientID) throws Exception { diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java index 940e310eed9d..2515d8220c4f 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java @@ -20,9 +20,11 @@ import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.ONE; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.when; import java.io.IOException; import java.util.ArrayList; @@ -40,6 +42,7 @@ import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; import org.apache.hadoop.ozone.om.response.OMClientResponse; +import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Part; @@ -236,6 +239,40 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( omClientResponse.getOMResponse().getStatus()); } + @Test + public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() + throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = getKeyName(); + + when(ozoneManager.getVersionManager() + .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT)).thenReturn(true); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + String multipartUploadID = + initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, + keyName, (byte) 1); + + OMRequest completeMultipartRequest = doPreExecuteCompleteMPU(volumeName, + bucketName, keyName, multipartUploadID, new ArrayList<>()); + + S3MultipartUploadCompleteRequest s3MultipartUploadCompleteRequest = + getS3MultipartUploadCompleteReq(completeMultipartRequest); + + OMClientResponse omClientResponse = + s3MultipartUploadCompleteRequest.validateAndUpdateCache(ozoneManager, + 3L); + + assertNotEquals(OzoneManagerProtocolProtos.Status + .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, + omClientResponse.getOMResponse().getStatus()); + assertEquals(OzoneManagerProtocolProtos.Status.INVALID_REQUEST, + omClientResponse.getOMResponse().getStatus()); + } + protected void addVolumeAndBucket(String volumeName, String bucketName) throws Exception { OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, From 001d61099678f7bdbc7e098e1b2ddbb13d370a19 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Tue, 9 Jun 2026 11:07:50 +0530 Subject: [PATCH 07/11] Address review comments, fix gate check --- .../ozone/om/helpers/OmMultipartKeyInfo.java | 15 +++++++-------- .../multipart/S3MultipartUploadAbortRequest.java | 11 +++++------ .../S3MultipartUploadCommitPartRequest.java | 14 ++++++++------ .../S3MultipartUploadCompleteRequest.java | 11 +++++------ .../TestS3MultipartUploadAbortRequest.java | 4 ++-- .../TestS3MultipartUploadCommitPartRequest.java | 4 ++-- .../TestS3MultipartUploadCompleteRequest.java | 4 ++-- 7 files changed, 31 insertions(+), 32 deletions(-) diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java index 8802d4eb0c00..c9fafc989e17 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java @@ -89,7 +89,7 @@ public final class OmMultipartKeyInfo extends WithObjectID implements CopyObject // 1 - New Schema -> Uses a separate table to store the multipart part info private static final int LEGACY_SCHEMA_VERSION = 0; private static final int SPLIT_PARTS_SCHEMA_VERSION = 1; - private final byte schemaVersion; + private final int schemaVersion; public static Codec getCodec() { return CODEC; @@ -275,7 +275,7 @@ public ReplicationConfig getReplicationConfig() { return replicationConfig; } - public byte getSchemaVersion() { + public int getSchemaVersion() { return schemaVersion; } @@ -297,7 +297,7 @@ public static class Builder extends WithObjectID.Builder { private final AclListBuilder acls; private final TreeMap partKeyInfoList; private long parentID; - private byte schemaVersion; + private int schemaVersion; public Builder() { this.acls = AclListBuilder.empty(); @@ -410,9 +410,8 @@ public Builder setParentID(long parentObjId) { return this; } - public Builder setSchemaVersion(byte schemaVersion) { - this.schemaVersion = - validateAndConvertSchemaVersion(Byte.toUnsignedInt(schemaVersion)); + public Builder setSchemaVersion(int schemaVersion) { + this.schemaVersion = validateAndConvertSchemaVersion(schemaVersion); return this; } @@ -520,13 +519,13 @@ public MultipartKeyInfo getProto() { return builder.build(); } - private static byte validateAndConvertSchemaVersion(int schemaVersion) { + private static int validateAndConvertSchemaVersion(int schemaVersion) { if (schemaVersion != LEGACY_SCHEMA_VERSION && schemaVersion != SPLIT_PARTS_SCHEMA_VERSION) { throw new IllegalArgumentException("Unsupported schemaVersion: " + schemaVersion + ". Expected one of [0, 1]."); } - return (byte) schemaVersion; + return schemaVersion; } @Override diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java index d8ecfd014f5d..8ec06141a815 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java @@ -165,12 +165,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - // This gate runs in the replicated apply path, so the check must remain - // deterministic across replicas for a given log index. MLV advances only - // via the Ratis-logged finalize-upgrade request. - if (!ozoneManager.getVersionManager() - .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) - && multipartKeyInfo.getSchemaVersion() != 0) { + // Use the layout version stamped by the leader in preExecute so all + // replicas evaluate the gate deterministically at apply. + long requestLayoutVersion = getOmRequest().getLayoutVersion().getVersion(); + if (requestLayoutVersion < OMLayoutFeature.MPU_PARTS_TABLE_SPLIT + .layoutVersion() && multipartKeyInfo.getSchemaVersion() != 0) { throw new OMException("MPU parts-table split behavior is not allowed " + "before cluster finalization.", OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java index 2f52c0e5c16b..8326bec9e58c 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java @@ -173,17 +173,19 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - // This gate runs in the replicated apply path, so the check must remain - // deterministic across replicas for a given log index. MLV advances only - // via the Ratis-logged finalize-upgrade request. - if (!ozoneManager.getVersionManager() - .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) + // Use the layout version stamped by the leader in preExecute so all + // replicas evaluate the gate deterministically at apply. + long requestLayoutVersion = getOmRequest().getLayoutVersion().getVersion(); + boolean splitPartsFeatureAllowed = requestLayoutVersion + >= OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion(); + if (!splitPartsFeatureAllowed && multipartKeyInfo.getSchemaVersion() != 0) { throw new OMException("MPU parts-table split behavior is not allowed " + "before cluster finalization for commit part request.", OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); } - if (multipartKeyInfo.getSchemaVersion() == 1) { + if (splitPartsFeatureAllowed + && multipartKeyInfo.getSchemaVersion() == 1) { throw new OMException("MPU parts-table split commit path is not " + "supported in this write flow.", OMException.ResultCodes.NOT_SUPPORTED_OPERATION); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java index 5f2708b0f3f1..afc7ea9ff171 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java @@ -269,12 +269,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - // This gate runs in the replicated apply path, so the check must remain - // deterministic across replicas for a given log index. MLV advances only - // via the Ratis-logged finalize-upgrade request. - if (!ozoneManager.getVersionManager() - .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT) - && multipartKeyInfo.getSchemaVersion() != 0) { + // Use the layout version stamped by the leader in preExecute so all + // replicas evaluate the gate deterministically at apply. + long requestLayoutVersion = getOmRequest().getLayoutVersion().getVersion(); + if (requestLayoutVersion < OMLayoutFeature.MPU_PARTS_TABLE_SPLIT + .layoutVersion() && multipartKeyInfo.getSchemaVersion() != 0) { throw new OMException("MPU parts-table split behavior is not allowed " + "before cluster finalization for complete request.", OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java index aca99c6847ff..c1695f251ebb 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java @@ -137,8 +137,8 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); - when(ozoneManager.getVersionManager() - .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT)).thenReturn(true); + when(ozoneManager.getVersionManager().getMetadataLayoutVersion()) + .thenReturn(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion()); OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout()); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java index 5d4ef02c1a95..8bc00b6f5efd 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java @@ -174,8 +174,8 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); - when(ozoneManager.getVersionManager() - .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT)).thenReturn(true); + when(ozoneManager.getVersionManager().getMetadataLayoutVersion()) + .thenReturn(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion()); OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout()); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java index 2515d8220c4f..f676a3057314 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java @@ -246,8 +246,8 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); - when(ozoneManager.getVersionManager() - .isAllowed(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT)).thenReturn(true); + when(ozoneManager.getVersionManager().getMetadataLayoutVersion()) + .thenReturn(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion()); OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout()); From 7bfa71ee4e7229e7f7235fea1317a090e6eb60fc Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Wed, 10 Jun 2026 00:12:56 +0530 Subject: [PATCH 08/11] add schemaVersion consumption --- .../S3InitiateMultipartUploadRequest.java | 7 ++ ...InitiateMultipartUploadRequestWithFSO.java | 1 + .../TestS3InitiateMultipartUploadRequest.java | 64 +++++++++++++++++++ .../TestS3MultipartUploadAbortRequest.java | 27 +++++++- ...estS3MultipartUploadCommitPartRequest.java | 26 +++++++- .../TestS3MultipartUploadCompleteRequest.java | 20 +++++- 6 files changed, 136 insertions(+), 9 deletions(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java index 22f470c80c7b..2419924c4e71 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java @@ -195,6 +195,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut replicationConfig) .setObjectID(objectID) .setUpdateID(transactionLogIndex) + .setSchemaVersion(resolveMultipartSchemaVersion()) .build(); omKeyInfo = new OmKeyInfo.Builder() @@ -284,6 +285,12 @@ protected void logResult(OzoneManager ozoneManager, } } + protected int resolveMultipartSchemaVersion() { + long requestLayoutVersion = getOmRequest().getLayoutVersion().getVersion(); + return requestLayoutVersion >= OMLayoutFeature.MPU_PARTS_TABLE_SPLIT + .layoutVersion() ? 1 : 0; + } + @RequestFeatureValidator( conditions = ValidationCondition.CLUSTER_NEEDS_FINALIZATION, processingPhase = RequestProcessingPhase.PRE_PROCESS, diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequestWithFSO.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequestWithFSO.java index 919491d70499..64710838b82a 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequestWithFSO.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequestWithFSO.java @@ -168,6 +168,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut .setObjectID(pathInfoFSO.getLeafNodeObjectId()) .setUpdateID(transactionLogIndex) .setParentID(pathInfoFSO.getLastKnownParentId()) + .setSchemaVersion(resolveMultipartSchemaVersion()) .build(); omKeyInfo = new OmKeyInfo.Builder() diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3InitiateMultipartUploadRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3InitiateMultipartUploadRequest.java index 742d17a87b06..69a19177bb32 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3InitiateMultipartUploadRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3InitiateMultipartUploadRequest.java @@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.when; import java.util.ArrayList; import java.util.HashMap; @@ -32,8 +33,10 @@ import org.apache.hadoop.ozone.OzoneAcl; import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; +import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; import org.apache.hadoop.ozone.om.response.OMClientResponse; +import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; import org.junit.jupiter.api.Test; @@ -113,6 +116,67 @@ public void testValidateAndUpdateCache() throws Exception { } + @Test + public void testValidateAndUpdateCacheSetsSchemaVersionZeroBeforeFinalization() + throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = UUID.randomUUID().toString(); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + OMRequest modifiedRequest = doPreExecuteInitiateMPU(volumeName, + bucketName, keyName); + S3InitiateMultipartUploadRequest request = + getS3InitiateMultipartUploadReq(modifiedRequest); + + OMClientResponse response = + request.validateAndUpdateCache(ozoneManager, 100L); + assertEquals(OzoneManagerProtocolProtos.Status.OK, + response.getOMResponse().getStatus()); + + String multipartKey = getMultipartKey(volumeName, bucketName, keyName, + modifiedRequest.getInitiateMultiPartUploadRequest() + .getKeyArgs().getMultipartUploadID()); + OmMultipartKeyInfo multipartKeyInfo = omMetadataManager + .getMultipartInfoTable().get(multipartKey); + assertNotNull(multipartKeyInfo); + assertEquals(0, multipartKeyInfo.getSchemaVersion()); + } + + @Test + public void testValidateAndUpdateCacheSetsSchemaVersionOneAfterFinalization() + throws Exception { + String volumeName = UUID.randomUUID().toString(); + String bucketName = UUID.randomUUID().toString(); + String keyName = UUID.randomUUID().toString(); + + when(ozoneManager.getVersionManager().getMetadataLayoutVersion()) + .thenReturn(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion()); + + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, + omMetadataManager, getBucketLayout()); + + OMRequest modifiedRequest = doPreExecuteInitiateMPU(volumeName, + bucketName, keyName); + S3InitiateMultipartUploadRequest request = + getS3InitiateMultipartUploadReq(modifiedRequest); + + OMClientResponse response = + request.validateAndUpdateCache(ozoneManager, 100L); + assertEquals(OzoneManagerProtocolProtos.Status.OK, + response.getOMResponse().getStatus()); + + String multipartKey = getMultipartKey(volumeName, bucketName, keyName, + modifiedRequest.getInitiateMultiPartUploadRequest() + .getKeyArgs().getMultipartUploadID()); + OmMultipartKeyInfo multipartKeyInfo = omMetadataManager + .getMultipartInfoTable().get(multipartKey); + assertNotNull(multipartKeyInfo); + assertEquals(1, multipartKeyInfo.getSchemaVersion()); + } + @Test public void testValidateAndUpdateCacheWithBucketNotFound() throws Exception { String volumeName = UUID.randomUUID().toString(); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java index c1695f251ebb..cdb0dbb1315c 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java @@ -27,6 +27,7 @@ import java.util.UUID; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; import org.apache.hadoop.hdds.utils.db.cache.CacheValue; +import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; import org.apache.hadoop.ozone.om.response.OMClientResponse; import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature; @@ -114,6 +115,13 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, keyName, (byte) 1); + String multipartKey = omMetadataManager.getMultipartKey(volumeName, + bucketName, keyName, multipartUploadID); + OmMultipartKeyInfo multipartKeyInfo = omMetadataManager + .getMultipartInfoTable().get(multipartKey); + assertNotNull(multipartKeyInfo); + assertEquals(1, multipartKeyInfo.getSchemaVersion()); + OMRequest abortMPURequest = doPreExecuteAbortMPU(volumeName, bucketName, keyName, multipartUploadID); @@ -145,9 +153,22 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() createParentPath(volumeName, bucketName); - String multipartUploadID = - initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, - keyName, (byte) 1); + OMRequest initiateMPURequest = doPreExecuteInitiateMPU(volumeName, + bucketName, keyName); + S3InitiateMultipartUploadRequest s3InitiateMultipartUploadRequest = + getS3InitiateMultipartUploadReq(initiateMPURequest); + OMClientResponse initiateResponse = + s3InitiateMultipartUploadRequest.validateAndUpdateCache(ozoneManager, + 1L); + String multipartUploadID = initiateResponse.getOMResponse() + .getInitiateMultiPartUploadResponse().getMultipartUploadID(); + + String multipartKey = omMetadataManager.getMultipartKey(volumeName, + bucketName, keyName, multipartUploadID); + OmMultipartKeyInfo multipartKeyInfo = omMetadataManager + .getMultipartInfoTable().get(multipartKey); + assertNotNull(multipartKeyInfo); + assertEquals(1, multipartKeyInfo.getSchemaVersion()); OMRequest abortMPURequest = doPreExecuteAbortMPU(volumeName, bucketName, keyName, diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java index 8bc00b6f5efd..8894ba047331 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java @@ -147,6 +147,13 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, keyName, (byte) 1); + String multipartKey = omMetadataManager.getMultipartKey(volumeName, + bucketName, keyName, multipartUploadID); + OmMultipartKeyInfo multipartKeyInfo = omMetadataManager + .getMultipartInfoTable().get(multipartKey); + assertNotNull(multipartKeyInfo); + assertEquals(1, multipartKeyInfo.getSchemaVersion()); + long clientID = Time.now(); OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName, bucketName, keyName, clientID, multipartUploadID, 1); @@ -182,9 +189,22 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() createParentPath(volumeName, bucketName); - String multipartUploadID = - initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, - keyName, (byte) 1); + OMRequest initiateMPURequest = doPreExecuteInitiateMPU(volumeName, + bucketName, keyName); + S3InitiateMultipartUploadRequest s3InitiateMultipartUploadRequest = + getS3InitiateMultipartUploadReq(initiateMPURequest); + OMClientResponse initiateResponse = + s3InitiateMultipartUploadRequest.validateAndUpdateCache(ozoneManager, + 1L); + String multipartUploadID = initiateResponse.getOMResponse() + .getInitiateMultiPartUploadResponse().getMultipartUploadID(); + + String multipartKey = omMetadataManager.getMultipartKey(volumeName, + bucketName, keyName, multipartUploadID); + OmMultipartKeyInfo multipartKeyInfo = omMetadataManager + .getMultipartInfoTable().get(multipartKey); + assertNotNull(multipartKeyInfo); + assertEquals(1, multipartKeyInfo.getSchemaVersion()); long clientID = Time.now(); OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName, diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java index f676a3057314..22f9cb9f94b5 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java @@ -39,6 +39,7 @@ import org.apache.hadoop.ozone.OzoneConsts; import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; +import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo; import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; import org.apache.hadoop.ozone.om.response.OMClientResponse; @@ -252,9 +253,22 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout()); - String multipartUploadID = - initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, - keyName, (byte) 1); + OMRequest initiateMPURequest = doPreExecuteInitiateMPU(volumeName, + bucketName, keyName); + S3InitiateMultipartUploadRequest s3InitiateMultipartUploadRequest = + getS3InitiateMultipartUploadReq(initiateMPURequest); + OMClientResponse initiateResponse = + s3InitiateMultipartUploadRequest.validateAndUpdateCache(ozoneManager, + 1L); + String multipartUploadID = initiateResponse.getOMResponse() + .getInitiateMultiPartUploadResponse().getMultipartUploadID(); + + String multipartKey = getMultipartKey(volumeName, bucketName, keyName, + multipartUploadID); + OmMultipartKeyInfo multipartKeyInfo = omMetadataManager + .getMultipartInfoTable().get(multipartKey); + assertNotNull(multipartKeyInfo); + assertEquals(1, multipartKeyInfo.getSchemaVersion()); OMRequest completeMultipartRequest = doPreExecuteCompleteMPU(volumeName, bucketName, keyName, multipartUploadID, new ArrayList<>()); From 8415aca1ec9a68f1e3a7ff8560027287cebc3100 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Wed, 10 Jun 2026 01:38:54 +0530 Subject: [PATCH 09/11] Pin schemaVersion to 0 --- .../ozone/om/helpers/OmMultipartKeyInfo.java | 33 ++++++------------- .../om/helpers/TestOmMultipartKeyInfo.java | 17 ++++++---- .../S3InitiateMultipartUploadRequest.java | 8 +++-- .../S3MultipartUploadCommitPartRequest.java | 6 ---- .../TestS3InitiateMultipartUploadRequest.java | 4 +-- .../TestS3MultipartUploadAbortRequest.java | 4 +-- ...estS3MultipartUploadCommitPartRequest.java | 6 ++-- .../TestS3MultipartUploadCompleteRequest.java | 4 +-- 8 files changed, 34 insertions(+), 48 deletions(-) diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java index c9fafc989e17..0d2f4fe809fc 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java @@ -260,10 +260,6 @@ public PartKeyInfoMap getPartKeyInfoMap() { } public void addPartKeyInfo(PartKeyInfo partKeyInfo) { - if (schemaVersion == SPLIT_PARTS_SCHEMA_VERSION) { - throw new IllegalStateException( - "PartKeyInfoMap is not supported for schemaVersion 1"); - } this.partKeyInfoMap = PartKeyInfoMap.put(partKeyInfo, partKeyInfoMap); } @@ -316,10 +312,8 @@ public Builder(OmMultipartKeyInfo multipartKeyInfo) { this.acls = AclListBuilder.of(multipartKeyInfo.acls); this.partKeyInfoList = new TreeMap<>(); - if (multipartKeyInfo.getSchemaVersion() == LEGACY_SCHEMA_VERSION) { - for (PartKeyInfo partKeyInfo : multipartKeyInfo.partKeyInfoMap) { - this.partKeyInfoList.put(partKeyInfo.getPartNumber(), partKeyInfo); - } + for (PartKeyInfo partKeyInfo : multipartKeyInfo.partKeyInfoMap) { + this.partKeyInfoList.put(partKeyInfo.getPartNumber(), partKeyInfo); } this.parentID = multipartKeyInfo.parentID; @@ -429,11 +423,11 @@ protected OmMultipartKeyInfo buildObject() { public static Builder builderFromProto( MultipartKeyInfo multipartKeyInfo) { final SortedMap list = new TreeMap<>(); - if (!multipartKeyInfo.hasSchemaVersion() - || multipartKeyInfo.getSchemaVersion() == LEGACY_SCHEMA_VERSION) { - multipartKeyInfo.getPartKeyInfoListList().forEach(partKeyInfo -> - list.put(partKeyInfo.getPartNumber(), partKeyInfo)); - } + // Keep reading the embedded part list for both schema versions so MPU + // commit/complete/list flows stay backward compatible until the parts + // split-table write/read path is fully wired in all requests. + multipartKeyInfo.getPartKeyInfoListList().forEach(partKeyInfo -> + list.put(partKeyInfo.getPartNumber(), partKeyInfo)); final ReplicationConfig replicationConfig = ReplicationConfig.fromProto( multipartKeyInfo.getType(), @@ -477,13 +471,6 @@ public static OmMultipartKeyInfo getFromProto( * @return MultipartKeyInfo */ public MultipartKeyInfo getProto() { - if (schemaVersion == SPLIT_PARTS_SCHEMA_VERSION - && partKeyInfoMap != null - && partKeyInfoMap.size() > 0) { - throw new IllegalStateException( - "PartKeyInfoMap must be empty for schemaVersion 1"); - } - MultipartKeyInfo.Builder builder = MultipartKeyInfo.newBuilder() .setUploadID(uploadID) .setCreationTime(creationTime) @@ -513,9 +500,9 @@ public MultipartKeyInfo getProto() { } builder.addAllAcls(OzoneAclUtil.toProtobuf(acls)); - if (schemaVersion == LEGACY_SCHEMA_VERSION) { - builder.addAllPartKeyInfoList(partKeyInfoMap); - } + // Keep serializing the embedded part list for both schema versions for + // compatibility with existing MPU request flows. + builder.addAllPartKeyInfoList(partKeyInfoMap); return builder.build(); } diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java index ee8d7de50a73..661ccaa1077d 100644 --- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java +++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmMultipartKeyInfo.java @@ -118,17 +118,17 @@ public void distinctListOfParts() { } @Test - public void addPartKeyInfoRejectsSchemaVersionOne() { + public void addPartKeyInfoSupportsSchemaVersionOne() { OmMultipartKeyInfo subject = createSubject() - .setSchemaVersion((byte) 1) + .setSchemaVersion(1) .build(); - assertThrows(IllegalStateException.class, - () -> subject.addPartKeyInfo(createPart(createKeyInfo()).build())); + subject.addPartKeyInfo(createPart(createKeyInfo()).build()); + assertEquals(1, subject.getPartKeyInfoMap().size()); } @Test - public void getProtoRejectsLegacyPartListForSchemaVersionOne() { + public void getProtoSupportsPartListForSchemaVersionOne() { PartKeyInfo part = createPart(createKeyInfo()).build(); TreeMap legacyMap = new TreeMap<>(); legacyMap.put(part.getPartNumber(), part); @@ -136,13 +136,16 @@ public void getProtoRejectsLegacyPartListForSchemaVersionOne() { OmMultipartKeyInfo subject = new OmMultipartKeyInfo.Builder() .setUploadID(UUID.randomUUID().toString()) .setCreationTime(Time.now()) - .setSchemaVersion((byte) 1) + .setSchemaVersion(1) .setReplicationConfig(StandaloneReplicationConfig.getInstance( HddsProtos.ReplicationFactor.ONE)) .setPartKeyInfoList(legacyMap) .build(); - assertThrows(IllegalStateException.class, subject::getProto); + OzoneManagerProtocolProtos.MultipartKeyInfo proto = subject.getProto(); + OmMultipartKeyInfo fromProto = OmMultipartKeyInfo.getFromProto(proto); + assertEquals(1, fromProto.getSchemaVersion()); + assertEquals(1, fromProto.getPartKeyInfoMap().size()); } @Test diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java index 2419924c4e71..7d880e711320 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java @@ -286,9 +286,11 @@ protected void logResult(OzoneManager ozoneManager, } protected int resolveMultipartSchemaVersion() { - long requestLayoutVersion = getOmRequest().getLayoutVersion().getVersion(); - return requestLayoutVersion >= OMLayoutFeature.MPU_PARTS_TABLE_SPLIT - .layoutVersion() ? 1 : 0; + // Keep newly initiated MPUs on legacy schema until the complete split + // parts-table write/read paths are implemented across commit/list/complete. + // This method is intentionally kept as a single switch point so we can + // restore dynamic schema selection in a follow-up change. + return 0; } @RequestFeatureValidator( diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java index 8326bec9e58c..9d46b3513b46 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java @@ -184,12 +184,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut "before cluster finalization for commit part request.", OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); } - if (splitPartsFeatureAllowed - && multipartKeyInfo.getSchemaVersion() == 1) { - throw new OMException("MPU parts-table split commit path is not " + - "supported in this write flow.", - OMException.ResultCodes.NOT_SUPPORTED_OPERATION); - } // Add/Update user defined metadata. // Set the UpdateID to current transactionLogIndex omKeyInfo = omKeyInfo.toBuilder() diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3InitiateMultipartUploadRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3InitiateMultipartUploadRequest.java index 69a19177bb32..58435f50230e 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3InitiateMultipartUploadRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3InitiateMultipartUploadRequest.java @@ -146,7 +146,7 @@ public void testValidateAndUpdateCacheSetsSchemaVersionZeroBeforeFinalization() } @Test - public void testValidateAndUpdateCacheSetsSchemaVersionOneAfterFinalization() + public void testValidateAndUpdateCacheKeepsSchemaVersionZeroAfterFinalization() throws Exception { String volumeName = UUID.randomUUID().toString(); String bucketName = UUID.randomUUID().toString(); @@ -174,7 +174,7 @@ public void testValidateAndUpdateCacheSetsSchemaVersionOneAfterFinalization() OmMultipartKeyInfo multipartKeyInfo = omMetadataManager .getMultipartInfoTable().get(multipartKey); assertNotNull(multipartKeyInfo); - assertEquals(1, multipartKeyInfo.getSchemaVersion()); + assertEquals(0, multipartKeyInfo.getSchemaVersion()); } @Test diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java index cdb0dbb1315c..4dec83bf2d39 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java @@ -139,7 +139,7 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( } @Test - public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() + public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() throws Exception { String volumeName = UUID.randomUUID().toString(); String bucketName = UUID.randomUUID().toString(); @@ -168,7 +168,7 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() OmMultipartKeyInfo multipartKeyInfo = omMetadataManager .getMultipartInfoTable().get(multipartKey); assertNotNull(multipartKeyInfo); - assertEquals(1, multipartKeyInfo.getSchemaVersion()); + assertEquals(0, multipartKeyInfo.getSchemaVersion()); OMRequest abortMPURequest = doPreExecuteAbortMPU(volumeName, bucketName, keyName, diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java index 8894ba047331..5e0b12d92985 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java @@ -175,7 +175,7 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( } @Test - public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() + public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() throws Exception { String volumeName = UUID.randomUUID().toString(); String bucketName = UUID.randomUUID().toString(); @@ -204,7 +204,7 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() OmMultipartKeyInfo multipartKeyInfo = omMetadataManager .getMultipartInfoTable().get(multipartKey); assertNotNull(multipartKeyInfo); - assertEquals(1, multipartKeyInfo.getSchemaVersion()); + assertEquals(0, multipartKeyInfo.getSchemaVersion()); long clientID = Time.now(); OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName, @@ -222,7 +222,7 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() assertNotEquals(OzoneManagerProtocolProtos.Status .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, omClientResponse.getOMResponse().getStatus()); - assertEquals(OzoneManagerProtocolProtos.Status.NOT_SUPPORTED_OPERATION, + assertEquals(OzoneManagerProtocolProtos.Status.OK, omClientResponse.getOMResponse().getStatus()); } diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java index 22f9cb9f94b5..d65ac47e2980 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java @@ -241,7 +241,7 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( } @Test - public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() + public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() throws Exception { String volumeName = UUID.randomUUID().toString(); String bucketName = UUID.randomUUID().toString(); @@ -268,7 +268,7 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionOneAfterFinalization() OmMultipartKeyInfo multipartKeyInfo = omMetadataManager .getMultipartInfoTable().get(multipartKey); assertNotNull(multipartKeyInfo); - assertEquals(1, multipartKeyInfo.getSchemaVersion()); + assertEquals(0, multipartKeyInfo.getSchemaVersion()); OMRequest completeMultipartRequest = doPreExecuteCompleteMPU(volumeName, bucketName, keyName, multipartUploadID, new ArrayList<>()); From 4217097fcb391985512c278014bf931050f7daaf Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Sun, 14 Jun 2026 15:37:47 +0530 Subject: [PATCH 10/11] Add pre-processor check for upgrades --- .../src/main/proto/OmClientProtocol.proto | 1 + .../S3InitiateMultipartUploadRequest.java | 45 +++++++++++++++---- ...InitiateMultipartUploadRequestWithFSO.java | 3 +- .../S3MultipartUploadAbortRequest.java | 10 ----- .../S3MultipartUploadCommitPartRequest.java | 11 ----- .../S3MultipartUploadCompleteRequest.java | 10 ----- .../validation/ValidationCondition.java | 13 ++++++ .../TestS3MultipartUploadAbortRequest.java | 7 +-- ...estS3MultipartUploadCommitPartRequest.java | 8 ++-- .../TestS3MultipartUploadCompleteRequest.java | 7 +-- 10 files changed, 64 insertions(+), 51 deletions(-) diff --git a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto index 22c71cc29852..4e30ed9f530c 100644 --- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto +++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto @@ -1715,6 +1715,7 @@ message ServiceInfo { message MultipartInfoInitiateRequest { required KeyArgs keyArgs = 1; + optional uint32 schemaVersion = 2; } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java index 7d880e711320..e6568477e8d7 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java @@ -72,6 +72,7 @@ public class S3InitiateMultipartUploadRequest extends OMKeyRequest { private static final Logger LOG = LoggerFactory.getLogger(S3InitiateMultipartUploadRequest.class); + private static final int LEGACY_MPU_SCHEMA_VERSION = 0; public S3InitiateMultipartUploadRequest(OMRequest omRequest, BucketLayout bucketLayout) { @@ -100,10 +101,14 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException { KeyArgs resolvedArgs = resolveBucketAndCheckKeyAcls(newKeyArgs.build(), ozoneManager, ACLType.CREATE); + int schemaVersion = resolveMultipartSchemaVersion(multipartInfoInitiateRequest); + MultipartInfoInitiateRequest.Builder requestBuilder = + multipartInfoInitiateRequest.toBuilder() + .setKeyArgs(resolvedArgs) + .setSchemaVersion(schemaVersion); return getOmRequest().toBuilder() .setUserInfo(getUserInfo()) - .setInitiateMultiPartUploadRequest( - multipartInfoInitiateRequest.toBuilder().setKeyArgs(resolvedArgs)) + .setInitiateMultiPartUploadRequest(requestBuilder) .build(); } @@ -195,7 +200,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut replicationConfig) .setObjectID(objectID) .setUpdateID(transactionLogIndex) - .setSchemaVersion(resolveMultipartSchemaVersion()) + .setSchemaVersion( + resolveMultipartSchemaVersion(multipartInfoInitiateRequest)) .build(); omKeyInfo = new OmKeyInfo.Builder() @@ -285,12 +291,33 @@ protected void logResult(OzoneManager ozoneManager, } } - protected int resolveMultipartSchemaVersion() { - // Keep newly initiated MPUs on legacy schema until the complete split - // parts-table write/read paths are implemented across commit/list/complete. - // This method is intentionally kept as a single switch point so we can - // restore dynamic schema selection in a follow-up change. - return 0; + protected int resolveMultipartSchemaVersion( + MultipartInfoInitiateRequest multipartInfoInitiateRequest) { + if (!multipartInfoInitiateRequest.hasSchemaVersion()) { + return LEGACY_MPU_SCHEMA_VERSION; + } + return (int) multipartInfoInitiateRequest.getSchemaVersion(); + } + + @RequestFeatureValidator( + conditions = { + ValidationCondition.CLUSTER_NEEDS_FINALIZATION, + ValidationCondition.CLUSTER_HAS_MPU_PARTS_TABLE_SPLIT + }, + processingPhase = RequestProcessingPhase.PRE_PROCESS, + requestType = Type.InitiateMultiPartUpload + ) + public static OMRequest setSchemaVersionOnInitiateMultipartUpload( + OMRequest req, ValidationContext ctx) { + MultipartInfoInitiateRequest initiateRequest = + req.getInitiateMultiPartUploadRequest(); + + // Keep newly initiated MPUs on legacy schema until split parts-table + // write/read paths are fully implemented. + return req.toBuilder() + .setInitiateMultiPartUploadRequest(initiateRequest.toBuilder() + .setSchemaVersion(LEGACY_MPU_SCHEMA_VERSION)) + .build(); } @RequestFeatureValidator( diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequestWithFSO.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequestWithFSO.java index 64710838b82a..eb4d21d608b2 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequestWithFSO.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequestWithFSO.java @@ -168,7 +168,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut .setObjectID(pathInfoFSO.getLeafNodeObjectId()) .setUpdateID(transactionLogIndex) .setParentID(pathInfoFSO.getLastKnownParentId()) - .setSchemaVersion(resolveMultipartSchemaVersion()) + .setSchemaVersion( + resolveMultipartSchemaVersion(multipartInfoInitiateRequest)) .build(); omKeyInfo = new OmKeyInfo.Builder() diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java index 8ec06141a815..a9aeff0ac5d1 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java @@ -165,16 +165,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - // Use the layout version stamped by the leader in preExecute so all - // replicas evaluate the gate deterministically at apply. - long requestLayoutVersion = getOmRequest().getLayoutVersion().getVersion(); - if (requestLayoutVersion < OMLayoutFeature.MPU_PARTS_TABLE_SPLIT - .layoutVersion() && multipartKeyInfo.getSchemaVersion() != 0) { - throw new OMException("MPU parts-table split behavior is not allowed " + - "before cluster finalization.", - OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); - } - multipartKeyInfo = multipartKeyInfo.toBuilder() .setUpdateID(trxnLogIndex) .build(); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java index 9d46b3513b46..3bd3abbe8b81 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java @@ -173,17 +173,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - // Use the layout version stamped by the leader in preExecute so all - // replicas evaluate the gate deterministically at apply. - long requestLayoutVersion = getOmRequest().getLayoutVersion().getVersion(); - boolean splitPartsFeatureAllowed = requestLayoutVersion - >= OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion(); - if (!splitPartsFeatureAllowed - && multipartKeyInfo.getSchemaVersion() != 0) { - throw new OMException("MPU parts-table split behavior is not allowed " + - "before cluster finalization for commit part request.", - OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); - } // Add/Update user defined metadata. // Set the UpdateID to current transactionLogIndex omKeyInfo = omKeyInfo.toBuilder() diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java index afc7ea9ff171..a692af65db38 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java @@ -269,16 +269,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR); } - // Use the layout version stamped by the leader in preExecute so all - // replicas evaluate the gate deterministically at apply. - long requestLayoutVersion = getOmRequest().getLayoutVersion().getVersion(); - if (requestLayoutVersion < OMLayoutFeature.MPU_PARTS_TABLE_SPLIT - .layoutVersion() && multipartKeyInfo.getSchemaVersion() != 0) { - throw new OMException("MPU parts-table split behavior is not allowed " + - "before cluster finalization for complete request.", - OMException.ResultCodes.NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION); - } - // Conditional write validation (If-None-Match / If-Match). // BUCKET_LOCK is held, so validation and commit are atomic. // Only 412 PreconditionFailed is possible; 409 Conflict cannot occur. diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/validation/ValidationCondition.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/validation/ValidationCondition.java index 87f2188ea1e2..cd4adbf0954c 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/validation/ValidationCondition.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/validation/ValidationCondition.java @@ -18,6 +18,7 @@ package org.apache.hadoop.ozone.om.request.validation; import org.apache.hadoop.ozone.ClientVersion; +import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; /** @@ -30,6 +31,18 @@ * the client. */ public enum ValidationCondition { + /** + * Classifies validations that apply once the MPU split-parts-table feature + * layout version has been finalized in the cluster. + */ + CLUSTER_HAS_MPU_PARTS_TABLE_SPLIT { + @Override + public boolean shouldApply(OMRequest req, ValidationContext ctx) { + return ctx.versionManager().getMetadataLayoutVersion() + >= OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion(); + } + }, + /** * Classifies validations that has to run after an upgrade until the cluster * is in a pre-finalized state. diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java index 4dec83bf2d39..f09c0fb065b1 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java @@ -100,7 +100,7 @@ public void testValidateAndUpdateCache() throws Exception { } @Test - public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization() + public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() throws Exception { String volumeName = UUID.randomUUID().toString(); String bucketName = UUID.randomUUID().toString(); @@ -129,13 +129,14 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( S3MultipartUploadAbortRequest s3MultipartUploadAbortRequest = getS3MultipartUploadAbortReq(abortMPURequest); - // The multipart metadata exists, but schema version 1 is not allowed yet. OMClientResponse omClientResponse = s3MultipartUploadAbortRequest.validateAndUpdateCache(ozoneManager, 2L); - assertEquals(OzoneManagerProtocolProtos.Status + assertNotEquals(OzoneManagerProtocolProtos.Status .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, omClientResponse.getOMResponse().getStatus()); + assertEquals(OzoneManagerProtocolProtos.Status.OK, + omClientResponse.getOMResponse().getStatus()); } @Test diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java index 5e0b12d92985..46d1436edb13 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java @@ -132,7 +132,7 @@ public void testValidateAndUpdateCacheSuccess() throws Exception { } @Test - public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization() + public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() throws Exception { String volumeName = UUID.randomUUID().toString(); String bucketName = UUID.randomUUID().toString(); @@ -163,15 +163,15 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( addKeyToOpenKeyTable(volumeName, bucketName, keyName, clientID); - // Regular part metadata is present; the upgrade gate should still reject - // this schema version before commit proceeds. OMClientResponse omClientResponse = s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, 2L); - assertEquals(OzoneManagerProtocolProtos.Status + assertNotEquals(OzoneManagerProtocolProtos.Status .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, omClientResponse.getOMResponse().getStatus()); + assertEquals(OzoneManagerProtocolProtos.Status.OK, + omClientResponse.getOMResponse().getStatus()); } @Test diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java index d65ac47e2980..665a20b225de 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java @@ -211,7 +211,7 @@ private String checkValidateAndUpdateCacheSuccess(String volumeName, } @Test - public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization() + public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() throws Exception { String volumeName = UUID.randomUUID().toString(); String bucketName = UUID.randomUUID().toString(); @@ -224,7 +224,6 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, keyName, (byte) 1); - // The request is still rejected before the empty part-list validation. OMRequest completeMultipartRequest = doPreExecuteCompleteMPU(volumeName, bucketName, keyName, multipartUploadID, new ArrayList<>()); @@ -235,9 +234,11 @@ public void testValidateAndUpdateCacheRejectsSchemaVersionOneBeforeFinalization( s3MultipartUploadCompleteRequest.validateAndUpdateCache(ozoneManager, 3L); - assertEquals(OzoneManagerProtocolProtos.Status + assertNotEquals(OzoneManagerProtocolProtos.Status .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, omClientResponse.getOMResponse().getStatus()); + assertEquals(OzoneManagerProtocolProtos.Status.INVALID_REQUEST, + omClientResponse.getOMResponse().getStatus()); } @Test From d48009e9d0b1c41f2087f4c66e82f60ad2b4544b Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Tue, 23 Jun 2026 10:17:01 +0530 Subject: [PATCH 11/11] Address review comments --- .../ozone/om/helpers/OmMultipartKeyInfo.java | 4 +-- .../S3InitiateMultipartUploadRequest.java | 10 ++----- .../validation/ValidationCondition.java | 13 -------- .../s3/multipart/TestS3MultipartRequest.java | 2 +- .../TestS3MultipartUploadAbortRequest.java | 11 ++----- ...estS3MultipartUploadCommitPartRequest.java | 25 ++-------------- ...ltipartUploadCommitPartRequestWithFSO.java | 30 ------------------- .../TestS3MultipartUploadCompleteRequest.java | 11 ++----- 8 files changed, 15 insertions(+), 91 deletions(-) diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java index 0d2f4fe809fc..db29582c9c5f 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmMultipartKeyInfo.java @@ -87,8 +87,8 @@ public final class OmMultipartKeyInfo extends WithObjectID implements CopyObject // This stores the schema version of the multipart key. // 0 - Legacy Schema -> Uses the same table to store the multipart part info // 1 - New Schema -> Uses a separate table to store the multipart part info - private static final int LEGACY_SCHEMA_VERSION = 0; - private static final int SPLIT_PARTS_SCHEMA_VERSION = 1; + public static final int LEGACY_SCHEMA_VERSION = 0; + public static final int SPLIT_PARTS_SCHEMA_VERSION = 1; private final int schemaVersion; public static Codec getCodec() { diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java index e6568477e8d7..d2128b2b39b5 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java @@ -72,7 +72,6 @@ public class S3InitiateMultipartUploadRequest extends OMKeyRequest { private static final Logger LOG = LoggerFactory.getLogger(S3InitiateMultipartUploadRequest.class); - private static final int LEGACY_MPU_SCHEMA_VERSION = 0; public S3InitiateMultipartUploadRequest(OMRequest omRequest, BucketLayout bucketLayout) { @@ -294,16 +293,13 @@ protected void logResult(OzoneManager ozoneManager, protected int resolveMultipartSchemaVersion( MultipartInfoInitiateRequest multipartInfoInitiateRequest) { if (!multipartInfoInitiateRequest.hasSchemaVersion()) { - return LEGACY_MPU_SCHEMA_VERSION; + return OmMultipartKeyInfo.LEGACY_SCHEMA_VERSION; } return (int) multipartInfoInitiateRequest.getSchemaVersion(); } @RequestFeatureValidator( - conditions = { - ValidationCondition.CLUSTER_NEEDS_FINALIZATION, - ValidationCondition.CLUSTER_HAS_MPU_PARTS_TABLE_SPLIT - }, + conditions = ValidationCondition.CLUSTER_NEEDS_FINALIZATION, processingPhase = RequestProcessingPhase.PRE_PROCESS, requestType = Type.InitiateMultiPartUpload ) @@ -316,7 +312,7 @@ public static OMRequest setSchemaVersionOnInitiateMultipartUpload( // write/read paths are fully implemented. return req.toBuilder() .setInitiateMultiPartUploadRequest(initiateRequest.toBuilder() - .setSchemaVersion(LEGACY_MPU_SCHEMA_VERSION)) + .setSchemaVersion(OmMultipartKeyInfo.LEGACY_SCHEMA_VERSION)) .build(); } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/validation/ValidationCondition.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/validation/ValidationCondition.java index cd4adbf0954c..87f2188ea1e2 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/validation/ValidationCondition.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/validation/ValidationCondition.java @@ -18,7 +18,6 @@ package org.apache.hadoop.ozone.om.request.validation; import org.apache.hadoop.ozone.ClientVersion; -import org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; /** @@ -31,18 +30,6 @@ * the client. */ public enum ValidationCondition { - /** - * Classifies validations that apply once the MPU split-parts-table feature - * layout version has been finalized in the cluster. - */ - CLUSTER_HAS_MPU_PARTS_TABLE_SPLIT { - @Override - public boolean shouldApply(OMRequest req, ValidationContext ctx) { - return ctx.versionManager().getMetadataLayoutVersion() - >= OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion(); - } - }, - /** * Classifies validations that has to run after an upgrade until the cluster * is in a pre-finalized state. diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartRequest.java index 6e8afb8afe4d..3ee32e0f4b55 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartRequest.java @@ -315,7 +315,7 @@ protected OMRequest doPreExecuteCompleteMPU( */ protected String initiateMultipartUploadWithSchemaVersion( String volumeName, String bucketName, String keyName, - byte schemaVersion) throws Exception { + int schemaVersion) throws Exception { OMRequest initiateMPURequest = doPreExecuteInitiateMPU(volumeName, bucketName, keyName); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java index f09c0fb065b1..92a268b27ea6 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadAbortRequest.java @@ -18,7 +18,6 @@ package org.apache.hadoop.ozone.om.request.s3.multipart; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.mockito.Mockito.when; @@ -106,6 +105,7 @@ public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); + // The base test fixture is pre-finalized by default. OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout()); @@ -113,7 +113,7 @@ public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() String multipartUploadID = initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, - keyName, (byte) 1); + keyName, 1); String multipartKey = omMetadataManager.getMultipartKey(volumeName, bucketName, keyName, multipartUploadID); @@ -132,9 +132,6 @@ public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() OMClientResponse omClientResponse = s3MultipartUploadAbortRequest.validateAndUpdateCache(ozoneManager, 2L); - assertNotEquals(OzoneManagerProtocolProtos.Status - .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, - omClientResponse.getOMResponse().getStatus()); assertEquals(OzoneManagerProtocolProtos.Status.OK, omClientResponse.getOMResponse().getStatus()); } @@ -146,6 +143,7 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); + // Tests must explicitly bump metadata layout version to simulate finalized OM. when(ozoneManager.getVersionManager().getMetadataLayoutVersion()) .thenReturn(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion()); @@ -183,9 +181,6 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() assertEquals(OzoneManagerProtocolProtos.Status.OK, omClientResponse.getOMResponse().getStatus()); - assertNotEquals(OzoneManagerProtocolProtos.Status - .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, - omClientResponse.getOMResponse().getStatus()); } @Test diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java index 46d1436edb13..862b6fd40606 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java @@ -34,7 +34,6 @@ import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; -import org.apache.hadoop.hdds.utils.db.BatchOperation; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo; @@ -138,6 +137,7 @@ public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); + // The base test fixture is pre-finalized by default. OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout()); @@ -145,7 +145,7 @@ public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() String multipartUploadID = initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, - keyName, (byte) 1); + keyName, 1); String multipartKey = omMetadataManager.getMultipartKey(volumeName, bucketName, keyName, multipartUploadID); @@ -167,9 +167,6 @@ public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, 2L); - assertNotEquals(OzoneManagerProtocolProtos.Status - .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, - omClientResponse.getOMResponse().getStatus()); assertEquals(OzoneManagerProtocolProtos.Status.OK, omClientResponse.getOMResponse().getStatus()); } @@ -181,6 +178,7 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); + // Tests must explicitly bump metadata layout version to simulate finalized OM. when(ozoneManager.getVersionManager().getMetadataLayoutVersion()) .thenReturn(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion()); @@ -219,9 +217,6 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, 2L); - assertNotEquals(OzoneManagerProtocolProtos.Status - .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, - omClientResponse.getOMResponse().getStatus()); assertEquals(OzoneManagerProtocolProtos.Status.OK, omClientResponse.getOMResponse().getStatus()); } @@ -248,10 +243,6 @@ public void testValidateAndUpdateCacheMultipartNotFound() throws Exception { // Add key to open key table. addKeyToOpenKeyTable(volumeName, bucketName, keyName, clientID); - String openKey = getOpenKey(volumeName, bucketName, keyName, clientID); - OmKeyInfo openPartKeyInfo = omMetadataManager.getOpenKeyTable( - s3MultipartUploadCommitPartRequest.getBucketLayout()).get(openKey); - assertNotNull(openPartKeyInfo); OMClientResponse omClientResponse = s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, 2L); @@ -263,16 +254,6 @@ public void testValidateAndUpdateCacheMultipartNotFound() throws Exception { bucketName, keyName, multipartUploadID); assertNull(omMetadataManager.getMultipartInfoTable().get(multipartKey)); - - BatchOperation batchOperation = omMetadataManager.getStore() - .initBatchOperation(); - omClientResponse.checkAndUpdateDB(omMetadataManager, batchOperation); - omMetadataManager.getStore().commitBatchOperation(batchOperation); - - String deleteKey = omMetadataManager.getOzoneDeletePathKey( - openPartKeyInfo.getObjectID(), multipartKey); - assertNotNull(omMetadataManager.getDeletedTable().get(deleteKey)); - } @Test diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequestWithFSO.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequestWithFSO.java index 6c823dcdaccb..4bff04d89138 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequestWithFSO.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequestWithFSO.java @@ -18,7 +18,6 @@ package org.apache.hadoop.ozone.om.request.s3.multipart; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -36,12 +35,8 @@ import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup; import org.apache.hadoop.ozone.om.helpers.OzoneFSUtils; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; -import org.apache.hadoop.ozone.om.response.OMClientResponse; -import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; import org.apache.hadoop.security.UserGroupInformation; -import org.apache.hadoop.util.Time; -import org.junit.jupiter.api.Test; /** * Tests S3 Multipart upload commit part request. @@ -76,31 +71,6 @@ protected String getKeyName() { return dirName + UUID.randomUUID().toString(); } - @Test - public void testValidateAndUpdateCacheDirectoryNotFound() throws Exception { - String volumeName = UUID.randomUUID().toString(); - String bucketName = UUID.randomUUID().toString(); - String keyName = getKeyName(); - - OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, - omMetadataManager, getBucketLayout()); - - long clientID = Time.now(); - String multipartUploadID = UUID.randomUUID().toString(); - OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName, - bucketName, keyName, clientID, multipartUploadID, 1); - - S3MultipartUploadCommitPartRequest s3MultipartUploadCommitPartRequest = - getS3MultipartUploadCommitReq(commitMultipartRequest); - - OMClientResponse omClientResponse = - s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, - 2L); - - assertEquals(OzoneManagerProtocolProtos.Status.DIRECTORY_NOT_FOUND, - omClientResponse.getOMResponse().getStatus()); - } - @Override protected void addKeyToOpenKeyTable(String volumeName, String bucketName, String keyName, long clientID) throws Exception { diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java index 665a20b225de..c4919147e982 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCompleteRequest.java @@ -20,7 +20,6 @@ import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.ONE; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -217,12 +216,13 @@ public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); + // The base test fixture is pre-finalized by default. OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout()); String multipartUploadID = initiateMultipartUploadWithSchemaVersion(volumeName, bucketName, - keyName, (byte) 1); + keyName, 1); OMRequest completeMultipartRequest = doPreExecuteCompleteMPU(volumeName, bucketName, keyName, multipartUploadID, new ArrayList<>()); @@ -234,9 +234,6 @@ public void testValidateAndUpdateCacheUsesSchemaVersionOneBeforeFinalization() s3MultipartUploadCompleteRequest.validateAndUpdateCache(ozoneManager, 3L); - assertNotEquals(OzoneManagerProtocolProtos.Status - .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, - omClientResponse.getOMResponse().getStatus()); assertEquals(OzoneManagerProtocolProtos.Status.INVALID_REQUEST, omClientResponse.getOMResponse().getStatus()); } @@ -248,6 +245,7 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() String bucketName = UUID.randomUUID().toString(); String keyName = getKeyName(); + // Tests must explicitly bump metadata layout version to simulate finalized OM. when(ozoneManager.getVersionManager().getMetadataLayoutVersion()) .thenReturn(OMLayoutFeature.MPU_PARTS_TABLE_SPLIT.layoutVersion()); @@ -281,9 +279,6 @@ public void testValidateAndUpdateCacheAllowsSchemaVersionZeroAfterFinalization() s3MultipartUploadCompleteRequest.validateAndUpdateCache(ozoneManager, 3L); - assertNotEquals(OzoneManagerProtocolProtos.Status - .NOT_SUPPORTED_OPERATION_PRIOR_FINALIZATION, - omClientResponse.getOMResponse().getStatus()); assertEquals(OzoneManagerProtocolProtos.Status.INVALID_REQUEST, omClientResponse.getOMResponse().getStatus()); }