diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java index 2d0b2bb56fd..d46cdeaf1fd 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java @@ -50,6 +50,9 @@ public enum OzoneManagerVersion implements ComponentVersion { S3_OBJECT_TAGGING_API(9, "OzoneManager version that supports S3 object tagging APIs, such as " + "PutObjectTagging, GetObjectTagging, and DeleteObjectTagging"), + S3_PART_AWARE_GET(10, "OzoneManager version that supports S3 get for a specific multipart " + + "upload part number"), + FUTURE_VERSION(-1, "Used internally in the client when the server side is " + " newer and an unknown server version has arrived to the client."); diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java index 93c675d9b90..2f28d66d7a7 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java @@ -1765,16 +1765,21 @@ public OzoneKeyDetails getS3KeyDetails(String bucketName, String keyName) @Override public OzoneKeyDetails getS3KeyDetails(String bucketName, String keyName, int partNumber) throws IOException { - OmKeyInfo keyInfo = getS3KeyInfo(bucketName, keyName, false); - List filteredKeyLocationInfo = keyInfo - .getLatestVersionLocations().getBlocksLatestVersionOnly().stream() - .filter(omKeyLocationInfo -> omKeyLocationInfo.getPartNumber() == - partNumber) - .collect(Collectors.toList()); - keyInfo.updateLocationInfoList(filteredKeyLocationInfo, false); - keyInfo.setDataSize(filteredKeyLocationInfo.stream() - .mapToLong(OmKeyLocationInfo::getLength) - .sum()); + OmKeyInfo keyInfo; + if (omVersion.compareTo(OzoneManagerVersion.S3_PART_AWARE_GET) >= 0) { + keyInfo = getS3PartKeyInfo(bucketName, keyName, partNumber); + } else { + keyInfo = getS3KeyInfo(bucketName, keyName, false); + List filteredKeyLocationInfo = keyInfo + .getLatestVersionLocations().getBlocksLatestVersionOnly().stream() + .filter(omKeyLocationInfo -> omKeyLocationInfo.getPartNumber() == + partNumber) + .collect(Collectors.toList()); + keyInfo.updateLocationInfoList(filteredKeyLocationInfo, true, true); + keyInfo.setDataSize(filteredKeyLocationInfo.stream() + .mapToLong(OmKeyLocationInfo::getLength) + .sum()); + } return getOzoneKeyDetails(keyInfo); } @@ -1801,6 +1806,29 @@ private OmKeyInfo getS3KeyInfo( return keyInfoWithS3Context.getKeyInfo(); } + @Nonnull + private OmKeyInfo getS3PartKeyInfo( + String bucketName, String keyName, int partNumber) throws IOException { + verifyBucketName(bucketName); + Preconditions.checkNotNull(keyName); + + OmKeyArgs keyArgs = new OmKeyArgs.Builder() + // Volume name is not important, as we call GetKeyInfo with + // assumeS3Context = true, OM will infer the correct s3 volume. + .setVolumeName(OzoneConfigKeys.OZONE_S3_VOLUME_NAME_DEFAULT) + .setBucketName(bucketName) + .setKeyName(keyName) + .setSortDatanodesInPipeline(topologyAwareReadEnabled) + .setLatestVersionLocation(getLatestVersionLocation) + .setForceUpdateContainerCacheFromSCM(false) + .setMultipartUploadPartNumber(partNumber) + .build(); + KeyInfoWithVolumeContext keyInfoWithS3Context = + ozoneManagerClient.getKeyInfo(keyArgs, true); + keyInfoWithS3Context.getUserPrincipal().ifPresent(this::updateS3Principal); + return keyInfoWithS3Context.getKeyInfo(); + } + private OmKeyInfo getKeyInfo( String volumeName, String bucketName, String keyName, boolean forceUpdateContainerCache) throws IOException { diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java index ba28b45a0e5..106ef6a06ab 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java @@ -213,6 +213,7 @@ public OmKeyArgs.Builder toBuilder() { if (expectedDataGeneration != null) { builder.setExpectedDataGeneration(expectedDataGeneration); } + return builder; } @@ -227,7 +228,11 @@ public KeyArgs toProtobuf() { .setLatestVersionLocation(getLatestVersionLocation()) .setHeadOp(isHeadOp()) .setForceUpdateContainerCacheFromSCM( - isForceUpdateContainerCacheFromSCM()); + isForceUpdateContainerCacheFromSCM() + ); + if (multipartUploadPartNumber != 0) { + builder.setMultipartNumber(multipartUploadPartNumber); + } if (expectedDataGeneration != null) { builder.setExpectedDataGeneration(expectedDataGeneration); } @@ -308,8 +313,8 @@ public Builder setMultipartUploadID(String uploadID) { return this; } - public Builder setMultipartUploadPartNumber(int partNumber) { - this.multipartUploadPartNumber = partNumber; + public Builder setMultipartUploadPartNumber(int multipartUploadPartNumber) { + this.multipartUploadPartNumber = multipartUploadPartNumber; return this; } diff --git a/hadoop-ozone/dist/src/main/smoketest/s3/objectputget.robot b/hadoop-ozone/dist/src/main/smoketest/s3/objectputget.robot index cd5a7c7597c..e204e177d8d 100644 --- a/hadoop-ozone/dist/src/main/smoketest/s3/objectputget.robot +++ b/hadoop-ozone/dist/src/main/smoketest/s3/objectputget.robot @@ -270,3 +270,36 @@ Create key twice with different content and expect different ETags Execute AWSS3Cli rm s3://${BUCKET}/test_key_to_check_etag_differences Execute rm -rf /tmp/file1 Execute rm -rf /tmp/file2 + +Create&Download big file by multipart upload and get file via part numbers + Execute head -c 10000000 /tmp/big_file + ${result} Execute AWSS3CliDebug cp /tmp/big_file s3://${BUCKET}/ + ${get_part_1_response} Execute AWSS3APICli get-object --bucket ${BUCKET} --key big_file /tmp/big_file_1 --part-number 1 + ${part_1_size} = Execute and checkrc echo '${get_part_1_response}' | jq -r '.ContentLength' 0 + Should contain ${get_part_1_response} \"PartsCount\": 2 + ${get_part_2_response} Execute AWSS3APICli get-object --bucket ${BUCKET} --key big_file /tmp/big_file_2 --part-number 2 + ${part_2_size} = Execute and checkrc echo '${get_part_2_response}' | jq -r '.ContentLength' 0 + Should contain ${get_part_2_response} \"PartsCount\": 2 + + Should Be Equal As Integers 10000000 ${${part_1_size} + ${part_2_size}} + + ${get_part_3_response} Execute AWSS3APICli get-object --bucket ${BUCKET} --key big_file /tmp/big_file_3 --part-number 3 + Should contain ${get_part_3_response} \"ContentLength\": 0 + Should contain ${get_part_3_response} \"PartsCount\": 2 + # clean up + Execute AWSS3Cli rm s3://${BUCKET}/big_file + Execute rm -rf /tmp/big_file + Execute rm -rf /tmp/big_file_1 + Execute rm -rf /tmp/big_file_2 + Execute rm -rf /tmp/big_file_3 + +Create&Download big file by multipart upload and get file not existed part number + Execute head -c 10000000 /tmp/big_file + ${result} Execute AWSS3CliDebug cp /tmp/big_file s3://${BUCKET}/ + ${get_part_99_response} Execute AWSS3APICli get-object --bucket ${BUCKET} --key big_file /tmp/big_file_1 --part-number 99 + Should contain ${get_part_99_response} \"ContentLength\": 0 + Should contain ${get_part_99_response} \"PartsCount\": 2 + # clean up + Execute AWSS3Cli rm s3://${BUCKET}/big_file + Execute rm -rf /tmp/big_file + Execute rm -rf /tmp/big_file_1 \ No newline at end of file diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientMultipartUploadWithFSO.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientMultipartUploadWithFSO.java index b943930f62f..58183e87705 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientMultipartUploadWithFSO.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientMultipartUploadWithFSO.java @@ -40,6 +40,8 @@ import org.apache.hadoop.ozone.client.OzoneBucket; import org.apache.hadoop.ozone.client.OzoneClient; import org.apache.hadoop.ozone.client.OzoneClientFactory; +import org.apache.hadoop.ozone.client.OzoneKeyDetails; +import org.apache.hadoop.ozone.client.OzoneKeyLocation; import org.apache.hadoop.ozone.client.OzoneMultipartUpload; import org.apache.hadoop.ozone.client.OzoneMultipartUploadList; import org.apache.hadoop.ozone.client.OzoneMultipartUploadPartListParts; @@ -179,12 +181,12 @@ public void preTest() throws Exception { @Test public void testInitiateMultipartUploadWithReplicationInformationSet() throws IOException { - String uploadID = initiateMultipartUpload(bucket, keyName, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, ReplicationType.RATIS, ONE); // Call initiate multipart upload for the same key again, this should // generate a new uploadID. - String uploadIDNew = initiateMultipartUpload(bucket, keyName, + String uploadIDNew = initiateMultipartUploadWithAsserts(bucket, keyName, ReplicationType.RATIS, ONE); assertNotEquals(uploadIDNew, uploadID); } @@ -216,7 +218,7 @@ public void testInitiateMultipartUploadWithDefaultReplication() throws @Test public void testUploadPartWithNoOverride() throws IOException { String sampleData = "sample Value"; - String uploadID = initiateMultipartUpload(bucket, keyName, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, ReplicationType.RATIS, ONE); OzoneOutputStream ozoneOutputStream = bucket.createMultipartKey(keyName, @@ -235,7 +237,7 @@ public void testUploadPartWithNoOverride() throws IOException { @Test public void testUploadPartOverrideWithRatis() throws Exception { String sampleData = "sample Value"; - String uploadID = initiateMultipartUpload(bucket, keyName, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, ReplicationType.RATIS, THREE); int partNumber = 1; @@ -348,7 +350,7 @@ private OzoneBucket getOzoneECBucket(String myBucket) @Test public void testMultipartUploadWithPartsLessThanMinSize() throws Exception { // Initiate multipart upload - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); // Upload Parts @@ -371,7 +373,7 @@ public void testMultipartUploadWithPartsLessThanMinSize() throws Exception { public void testMultipartUploadWithDiscardedUnusedPartSize() throws Exception { // Initiate multipart upload - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, ONE); + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); byte[] data = generateData(10000000, (byte) 97); // Upload Parts @@ -402,7 +404,7 @@ public void testMultipartUploadWithDiscardedUnusedPartSize() @Test public void testMultipartUploadWithPartsMisMatchWithListSizeDifferent() throws Exception { - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); // We have not uploaded any parts, but passing some list it should throw @@ -417,7 +419,7 @@ public void testMultipartUploadWithPartsMisMatchWithListSizeDifferent() @Test public void testMultipartUploadWithPartsMisMatchWithIncorrectPartName() throws Exception { - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); uploadPart(bucket, keyName, uploadID, 1, "data".getBytes(UTF_8)); @@ -432,7 +434,7 @@ public void testMultipartUploadWithPartsMisMatchWithIncorrectPartName() @Test public void testMultipartUploadWithMissingParts() throws Exception { - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); uploadPart(bucket, keyName, uploadID, 1, "data".getBytes(UTF_8)); @@ -447,7 +449,7 @@ public void testMultipartUploadWithMissingParts() throws Exception { @Test public void testMultipartPartNumberExceedingAllowedRange() throws Exception { - String uploadID = initiateMultipartUpload(bucket, keyName, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); byte[] data = "data".getBytes(UTF_8); @@ -469,7 +471,7 @@ public void testMultipartPartNumberExceedingAllowedRange() throws Exception { public void testCommitPartAfterCompleteUpload() throws Exception { String parentDir = "a/b/c/d/"; keyName = parentDir + UUID.randomUUID(); - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, ONE); + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); assertEquals(volume.getBucket(bucketName).getUsedNamespace(), 4); @@ -530,7 +532,7 @@ public void testAbortUploadFailWithInProgressPartUpload() throws Exception { String parentDir = "a/b/c/d/"; keyName = parentDir + UUID.randomUUID(); - String uploadID = initiateMultipartUpload(bucket, keyName, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); // Do not close output stream. @@ -550,7 +552,7 @@ public void testAbortUploadSuccessWithOutAnyParts() throws Exception { String parentDir = "a/b/c/d/"; keyName = parentDir + UUID.randomUUID(); - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); bucket.abortMultipartUpload(keyName, uploadID); } @@ -567,7 +569,7 @@ public void testAbortUploadSuccessWithParts() throws Exception { ozoneManager.getMetadataManager().getBucketTable().get(buckKey); BucketLayout bucketLayout = buckInfo.getBucketLayout(); - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); Pair partNameAndETag = uploadPart(bucket, keyName, uploadID, 1, "data".getBytes(UTF_8)); @@ -599,7 +601,7 @@ public void testListMultipartUploadParts() throws Exception { keyName = parentDir + "file-ABC"; Map partsMap = new TreeMap<>(); - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); Pair partNameAndETag1 = uploadPart(bucket, keyName, uploadID, 1, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte)97)); @@ -688,7 +690,7 @@ private String verifyPartNames(Map partsMap, int index, public void testListMultipartUploadPartsWithContinuation() throws Exception { Map partsMap = new TreeMap<>(); - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); Pair partNameAndETag1 = uploadPart(bucket, keyName, uploadID, 1, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte)97)); @@ -752,7 +754,7 @@ public void testListPartsWithInvalidInputs(int partNumberMarker, int maxParts, S @Test public void testListPartsWithPartMarkerGreaterThanPartCount() throws Exception { - String uploadID = initiateMultipartUpload(bucket, keyName, RATIS, + String uploadID = initiateMultipartUploadWithAsserts(bucket, keyName, RATIS, ONE); uploadPart(bucket, keyName, uploadID, 1, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte)97)); @@ -795,11 +797,11 @@ public void testListMultipartUpload() throws Exception { keys.add(key3); // Initiate multipart upload - String uploadID1 = initiateMultipartUpload(bucket, key1, RATIS, + String uploadID1 = initiateMultipartUploadWithAsserts(bucket, key1, RATIS, ONE); - String uploadID2 = initiateMultipartUpload(bucket, key2, RATIS, + String uploadID2 = initiateMultipartUploadWithAsserts(bucket, key2, RATIS, ONE); - String uploadID3 = initiateMultipartUpload(bucket, key3, RATIS, + String uploadID3 = initiateMultipartUploadWithAsserts(bucket, key3, RATIS, ONE); // Upload Parts @@ -854,6 +856,105 @@ public void testListMultipartUpload() throws Exception { assertEquals(0, expectedList.size()); } + @Test + void testGetAllPartsWhenZeroPartNumber() throws Exception { + String parentDir = "a/b/c/d/e/f/"; + keyName = parentDir + "file-ABC"; + OzoneVolume s3volume = store.getVolume("s3v"); + s3volume.createBucket(bucketName); + OzoneBucket s3Bucket = s3volume.getBucket(bucketName); + + Map partsMap = new TreeMap<>(); + String uploadID = initiateMultipartUpload(s3Bucket, keyName, RATIS, + ONE); + Pair partNameAndETag1 = uploadPart(s3Bucket, keyName, + uploadID, 1, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(1, partNameAndETag1.getKey()); + + Pair partNameAndETag2 = uploadPart(s3Bucket, keyName, + uploadID, 2, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(2, partNameAndETag2.getKey()); + + Pair partNameAndETag3 = uploadPart(s3Bucket, keyName, + uploadID, 3, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(3, partNameAndETag3.getKey()); + + s3Bucket.completeMultipartUpload(keyName, uploadID, partsMap); + + OzoneKeyDetails s3KeyDetailsWithAllParts = ozClient.getProxy() + .getS3KeyDetails(s3Bucket.getName(), keyName, 0); + List ozoneKeyLocations = s3KeyDetailsWithAllParts.getOzoneKeyLocations(); + assertEquals(6, ozoneKeyLocations.size()); + } + + @Test + void testGetParticularPart() throws Exception { + String parentDir = "a/b/c/d/e/f/"; + keyName = parentDir + "file-ABC"; + OzoneVolume s3volume = store.getVolume("s3v"); + s3volume.createBucket(bucketName); + OzoneBucket s3Bucket = s3volume.getBucket(bucketName); + + Map partsMap = new TreeMap<>(); + String uploadID = initiateMultipartUpload(s3Bucket, keyName, RATIS, + ONE); + Pair partNameAndETag1 = uploadPart(s3Bucket, keyName, + uploadID, 1, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(1, partNameAndETag1.getKey()); + + Pair partNameAndETag2 = uploadPart(s3Bucket, keyName, + uploadID, 2, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(2, partNameAndETag2.getKey()); + + Pair partNameAndETag3 = uploadPart(s3Bucket, keyName, + uploadID, 3, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(3, partNameAndETag3.getKey()); + + s3Bucket.completeMultipartUpload(keyName, uploadID, partsMap); + +// OzoneKeyLocations size is 2 because part size is 5MB and ozone.scm.block.size in ozone-site.xml +// for integration-test is 4MB + OzoneKeyDetails s3KeyDetailsOneParts = ozClient.getProxy().getS3KeyDetails(bucketName, keyName, 1); + assertEquals(2, s3KeyDetailsOneParts.getOzoneKeyLocations().size()); + + OzoneKeyDetails s3KeyDetailsTwoParts = ozClient.getProxy().getS3KeyDetails(bucketName, keyName, 2); + assertEquals(2, s3KeyDetailsTwoParts.getOzoneKeyLocations().size()); + + OzoneKeyDetails s3KeyDetailsThreeParts = ozClient.getProxy().getS3KeyDetails(bucketName, keyName, 3); + assertEquals(2, s3KeyDetailsThreeParts.getOzoneKeyLocations().size()); + } + + @Test + void testGetNotExistedPart() throws Exception { + String parentDir = "a/b/c/d/e/f/"; + keyName = parentDir + "file-ABC"; + OzoneVolume s3volume = store.getVolume("s3v"); + s3volume.createBucket(bucketName); + OzoneBucket s3Bucket = s3volume.getBucket(bucketName); + + Map partsMap = new TreeMap<>(); + String uploadID = initiateMultipartUpload(s3Bucket, keyName, RATIS, + ONE); + Pair partNameAndETag1 = uploadPart(s3Bucket, keyName, + uploadID, 1, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(1, partNameAndETag1.getKey()); + + Pair partNameAndETag2 = uploadPart(s3Bucket, keyName, + uploadID, 2, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(2, partNameAndETag2.getKey()); + + Pair partNameAndETag3 = uploadPart(s3Bucket, keyName, + uploadID, 3, generateData(OzoneConsts.OM_MULTIPART_MIN_SIZE, (byte) 97)); + partsMap.put(3, partNameAndETag3.getKey()); + + s3Bucket.completeMultipartUpload(keyName, uploadID, partsMap); + + OzoneKeyDetails s3KeyDetailsWithNotExistedParts = ozClient.getProxy() + .getS3KeyDetails(s3Bucket.getName(), keyName, 4); + List ozoneKeyLocations = s3KeyDetailsWithNotExistedParts.getOzoneKeyLocations(); + assertEquals(0, ozoneKeyLocations.size()); + } + private String verifyUploadedPart(String uploadID, String partName, OMMetadataManager metadataMgr) throws IOException { OzoneManager ozoneManager = cluster.getOzoneManager(); @@ -891,11 +992,10 @@ private String verifyUploadedPart(String uploadID, String partName, return multipartKey; } - private String initiateMultipartUpload(OzoneBucket oBucket, String kName, - ReplicationType replicationType, ReplicationFactor replicationFactor) - throws IOException { - OmMultipartInfo multipartInfo = oBucket.initiateMultipartUpload(kName, - replicationType, replicationFactor); + private String initiateMultipartUploadWithAsserts( + OzoneBucket oBucket, String kName, ReplicationType replicationType, ReplicationFactor replicationFactor + ) throws IOException { + OmMultipartInfo multipartInfo = oBucket.initiateMultipartUpload(kName, replicationType, replicationFactor); assertNotNull(multipartInfo); String uploadID = multipartInfo.getUploadID(); @@ -907,6 +1007,13 @@ private String initiateMultipartUpload(OzoneBucket oBucket, String kName, return uploadID; } + private String initiateMultipartUpload( + OzoneBucket oBucket, String kName, ReplicationType replicationType, ReplicationFactor replicationFactor + ) throws IOException { + OmMultipartInfo multipartInfo = oBucket.initiateMultipartUpload(kName, replicationType, replicationFactor); + return multipartInfo.getUploadID(); + } + private Pair uploadPart(OzoneBucket oBucket, String kName, String uploadID, int partNumber, byte[] data) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java index e9c9b946c8e..c3ceb0f6209 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java @@ -1518,7 +1518,107 @@ public void testRefreshPipelineException() throws Exception { assertEquals(errorMessage, omEx.getMessage()); } - /** + @Test + void testGetAllPartsWhenZeroPartNumber() throws IOException { + String keyName = RandomStringUtils.randomAlphabetic(5); + + String volume = VOLUME_NAME; + + initKeyTableForMultipartTest(keyName, volume); + + OmKeyArgs keyArgs = new OmKeyArgs.Builder() + .setVolumeName(volume) + .setBucketName(BUCKET_NAME) + .setKeyName(keyName) + .setMultipartUploadPartNumber(0) + .build(); + OmKeyInfo omKeyInfo = keyManager.getKeyInfo(keyArgs, resolvedBucket(), "test"); + assertEquals(keyName, omKeyInfo.getKeyName()); + assertNotNull(omKeyInfo.getLatestVersionLocations()); + + List locationList = omKeyInfo.getLatestVersionLocations().getLocationList(); + assertNotNull(locationList); + assertEquals(5, locationList.size()); + for (int i = 0; i < 5; i++) { + assertEquals(i, locationList.get(i).getPartNumber()); + } + } + + @Test + void testGetParticularPart() throws IOException { + String keyName = RandomStringUtils.randomAlphabetic(5); + + String volume = VOLUME_NAME; + + initKeyTableForMultipartTest(keyName, volume); + + OmKeyArgs keyArgs = new OmKeyArgs.Builder() + .setVolumeName(volume) + .setBucketName(BUCKET_NAME) + .setKeyName(keyName) + .setMultipartUploadPartNumber(3) + .build(); + OmKeyInfo omKeyInfo = keyManager.getKeyInfo(keyArgs, resolvedBucket(), "test"); + assertEquals(keyName, omKeyInfo.getKeyName()); + assertNotNull(omKeyInfo.getLatestVersionLocations()); + + List locationList = omKeyInfo.getLatestVersionLocations().getLocationList(); + assertNotNull(locationList); + assertEquals(1, locationList.size()); + assertEquals(3, locationList.get(0).getPartNumber()); + } + + @Test + void testGetNotExistedPart() throws IOException { + String keyName = RandomStringUtils.randomAlphabetic(5); + + String volume = VOLUME_NAME; + + initKeyTableForMultipartTest(keyName, volume); + + OmKeyArgs keyArgs = new OmKeyArgs.Builder() + .setVolumeName(volume) + .setBucketName(BUCKET_NAME) + .setKeyName(keyName) + .setMultipartUploadPartNumber(99) + .build(); + OmKeyInfo omKeyInfo = keyManager.getKeyInfo(keyArgs, resolvedBucket(), "test"); + assertEquals(keyName, omKeyInfo.getKeyName()); + assertNotNull(omKeyInfo.getLatestVersionLocations()); + + List locationList = omKeyInfo.getLatestVersionLocations().getLocationList(); + assertNotNull(locationList); + assertEquals(0, locationList.size()); + } + + private void initKeyTableForMultipartTest(String keyName, String volume) throws IOException { + List locationInfoGroups = new ArrayList<>(); + List locationInfoList = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + OmKeyLocationInfo locationInfo1 = new OmKeyLocationInfo.Builder() + .setBlockID(new BlockID(i, i)) + .setPartNumber(i) + .build(); + locationInfoList.add(locationInfo1); + } + + OmKeyLocationInfoGroup locationInfoGroup = new OmKeyLocationInfoGroup(0, locationInfoList); + locationInfoGroups.add(locationInfoGroup); + locationInfoGroup.setMultipartKey(true); + + OmKeyInfo omKeyInfo = new OmKeyInfo.Builder() + .setKeyName(keyName) + .setBucketName(BUCKET_NAME) + .setVolumeName(volume) + .setReplicationConfig(RatisReplicationConfig.getInstance(THREE)) + .setOmKeyLocationInfos(locationInfoGroups) + .build(); + + String key = String.format("/%s/%s/%s", volume, BUCKET_NAME, keyName); + metadataManager.getKeyTable(BucketLayout.LEGACY).put(key, omKeyInfo); + } + + /** * Get Random pipeline. * @return pipeline */ diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java index ab56af670b3..661afaf0772 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java @@ -27,6 +27,7 @@ import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest; import com.amazonaws.services.s3.model.CompleteMultipartUploadResult; import com.amazonaws.services.s3.model.CreateBucketRequest; +import com.amazonaws.services.s3.model.GetObjectRequest; import com.amazonaws.services.s3.model.Grantee; import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest; import com.amazonaws.services.s3.model.InitiateMultipartUploadResult; @@ -736,6 +737,65 @@ public void testListParts(@TempDir Path tempDir) throws Exception { } } + @Test + public void testGetParticularPart(@TempDir Path tempDir) throws Exception { + final String bucketName = getBucketName(); + final String keyName = getKeyName(); + + s3Client.createBucket(bucketName); + + File multipartUploadFile = Files.createFile(tempDir.resolve("multipartupload.txt")).toFile(); + + createFile(multipartUploadFile, (int) (15 * MB)); + + multipartUpload(bucketName, keyName, multipartUploadFile, 5 * MB, null, null, null); + + GetObjectRequest getObjectRequestAll = new GetObjectRequest(bucketName, keyName); + getObjectRequestAll.setPartNumber(0); + S3Object s3ObjectAll = s3Client.getObject(getObjectRequestAll); + long allPartContentLength = s3ObjectAll.getObjectMetadata().getContentLength(); + + GetObjectRequest getObjectRequestOne = new GetObjectRequest(bucketName, keyName); + getObjectRequestOne.setPartNumber(1); + S3Object s3ObjectOne = s3Client.getObject(getObjectRequestOne); + long partOneContentLength = s3ObjectOne.getObjectMetadata().getContentLength(); + assertEquals(allPartContentLength / 3, partOneContentLength); + + GetObjectRequest getObjectRequestTwo = new GetObjectRequest(bucketName, keyName); + getObjectRequestTwo.setPartNumber(2); + S3Object s3ObjectTwo = s3Client.getObject(getObjectRequestTwo); + long partTwoContentLength = s3ObjectTwo.getObjectMetadata().getContentLength(); + assertEquals(allPartContentLength / 3, partTwoContentLength); + + GetObjectRequest getObjectRequestThree = new GetObjectRequest(bucketName, keyName); + getObjectRequestThree.setPartNumber(1); + S3Object s3ObjectThree = s3Client.getObject(getObjectRequestTwo); + long partThreeContentLength = s3ObjectThree.getObjectMetadata().getContentLength(); + assertEquals(allPartContentLength / 3, partThreeContentLength); + + assertEquals(allPartContentLength, (partOneContentLength + partTwoContentLength + partThreeContentLength)); + } + + @Test + public void testGetNotExistedPart(@TempDir Path tempDir) throws Exception { + final String bucketName = getBucketName(); + final String keyName = getKeyName(); + + s3Client.createBucket(bucketName); + + File multipartUploadFile = Files.createFile(tempDir.resolve("multipartupload.txt")).toFile(); + + createFile(multipartUploadFile, (int) (15 * MB)); + + multipartUpload(bucketName, keyName, multipartUploadFile, 5 * MB, null, null, null); + + GetObjectRequest getObjectRequestOne = new GetObjectRequest(bucketName, keyName); + getObjectRequestOne.setPartNumber(4); + S3Object s3ObjectOne = s3Client.getObject(getObjectRequestOne); + long partOneContentLength = s3ObjectOne.getObjectMetadata().getContentLength(); + assertEquals(0, partOneContentLength); + } + @Test public void testListPartsNotFound() { final String bucketName = getBucketName(); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java index ccda21efc93..8e3bbb47c3c 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java @@ -509,6 +509,22 @@ private OmKeyInfo readKeyInfo(OmKeyArgs args, BucketLayout bucketLayout) if (args.getLatestVersionLocation()) { slimLocationVersion(value); } + int partNumberParam = args.getMultipartUploadPartNumber(); + if (partNumberParam > 0) { + OmKeyLocationInfoGroup latestLocationVersion = value.getLatestVersionLocations(); + if (latestLocationVersion != null && latestLocationVersion.isMultipartKey()) { + List currentLocations = latestLocationVersion.getBlocksLatestVersionOnly() + .stream() + .filter(it -> it.getPartNumber() == partNumberParam) + .collect(Collectors.toList()); + + value.updateLocationInfoList(currentLocations, true, true); + + value.setDataSize(currentLocations.stream() + .mapToLong(BlockLocationInfo::getLength) + .sum()); + } + } return value; } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java index 09865ace27a..a31783f29c4 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java @@ -660,6 +660,7 @@ private GetKeyInfoResponse getKeyInfo(GetKeyInfoRequest request, .setHeadOp(keyArgs.getHeadOp()) .setForceUpdateContainerCacheFromSCM( keyArgs.getForceUpdateContainerCacheFromSCM()) + .setMultipartUploadPartNumber(keyArgs.getMultipartNumber()) .build(); KeyInfoWithVolumeContext keyInfo = impl.getKeyInfo(omKeyArgs, request.getAssumeS3Context());