[SDK-649] Refine code and add test case.

xiaoxin-ren · xiaoxin-ren · commit a30f21ede6e1 · 2024-05-17T21:57:44.000+01:00
diff --git a/src/main/java/com/emc/object/s3/LargeFileUploader.java b/src/main/java/com/emc/object/s3/LargeFileUploader.java
@@ -83,6 +83,7 @@ public static String getMpuETag(List<MultipartPartETag> partETags) {
 
     private final String srcBucket;
     private final String srcKey;
+    private String sourceVersionId;
 
     private final InputStream stream;
     private final LargeFileMultipartSource multipartSource;
@@ -153,13 +154,12 @@ public LargeFileUploader(S3Client s3Client, String bucket, String key, LargeFile
      * from <code>srcBucket/srcKey</code> to <code>dstBucket/dstKey</code> without streaming
      * data between the client and ECS server.
      */
-    public LargeFileUploader(S3Client s3Client, String srcBucket, String srcKey, String dstBucket, String dstKey, long size) {
+    public LargeFileUploader(S3Client s3Client, String srcBucket, String srcKey, String dstBucket, String dstKey) {
         this.s3Client = s3Client;
         this.bucket = dstBucket;
         this.key = dstKey;
         this.srcBucket = srcBucket;
         this.srcKey = srcKey;
-        this.fullSize = size;
         this.multipartSource = null;
         this.stream = null;
     }
@@ -305,6 +305,8 @@ public void upload() {
 
         if (fullSize >= mpuThreshold)
             doMultipartUpload();
+        else if (srcKey != null && srcBucket != null)
+            doSingleCopy();
         else
             doSinglePut();
     }
@@ -314,6 +316,8 @@ private InputStream getSourceCompleteDataStream() throws IOException {
         if (multipartSource != null) {
             is = multipartSource.getCompleteDataStream();
         } else {
+            if (stream == null)
+                throw new IOException("null source stream");
             // only close the source stream if configured to do so
             is = new FilterInputStream(stream) {
                 @Override
@@ -359,6 +363,17 @@ public void doSinglePut() {
         }
     }
 
+    public void doSingleCopy() {
+        configure();
+        if (srcKey == null || srcBucket == null)
+            throw new IllegalArgumentException("must specify source bucket and key for the copy");
+        CopyObjectRequest copyObjectRequest = new CopyObjectRequest(srcBucket, srcKey, bucket, key)
+                .withSourceVersionId(sourceVersionId).withObjectMetadata(objectMetadata)
+                .withAcl(acl).withCannedAcl(cannedAcl);
+        CopyObjectResult result = s3Client.copyObject(copyObjectRequest);
+        eTag = result.getRawETag();
+    }
+
     /*
      * get a map of existing MPU parts from which we can resume an MPU. we can only resume an MPU if the existing
      * part sizes and count are exactly the same as configured in this LFU instance
@@ -593,8 +608,9 @@ protected void configure() {
             executorService = null;
             threads = 1;
         } else if (srcKey != null && srcBucket != null) {
+            // If resuming from copied parts, no need to verify the parts found in target
             if (resumeContext != null) resumeContext.setVerifyPartsFoundInTarget(false);
-            executorService = null;
+            fullSize = s3Client.getObjectMetadata(new GetObjectMetadataRequest(srcBucket, srcKey).withVersionId(sourceVersionId)).getContentLength();
         } else {
             throw new IllegalArgumentException("must specify a file, stream, or multipartSource to read");
         }
@@ -803,6 +819,14 @@ public void setAbortMpuOnFailure(boolean abortMpuOnFailure) {
         this.abortMpuOnFailure = abortMpuOnFailure;
     }
 
+    public String getSourceVersionId() {
+        return sourceVersionId;
+    }
+
+    public void setSourceVersionId(String sourceVersionId) {
+        this.sourceVersionId = sourceVersionId;
+    }
+
     public LargeFileUploader withObjectMetadata(S3ObjectMetadata objectMetadata) {
         setObjectMetadata(objectMetadata);
         return this;
@@ -864,6 +888,11 @@ public LargeFileUploader withAbortMpuOnFailure(boolean abortMpuOnFailure) {
         return this;
     }
 
+    public LargeFileUploader withSourceVersionId(String sourceVersionId) {
+        setSourceVersionId(sourceVersionId);
+        return this;
+    }
+
     private class CopyPartTask implements Callable<MultipartPartETag> {
         private final String uploadId;
         private final int partNumber;
@@ -888,7 +917,7 @@ public MultipartPartETag call() {
                         .withSourceRange(new Range(offset, offset + length - 1));
                 try {
                     CopyPartResult result = s3Client.copyPart(copyPartRequest);
-                    return new MultipartPartETag(result.getPartNumber(), result.getETag());
+                    return new MultipartPartETag(result.getPartNumber(), result.getRawETag());
                 } catch (S3Exception e) {
                     throw new RuntimeException(e);
                 }
diff --git a/src/test/java/com/emc/object/s3/LargeFileUploaderTest.java b/src/test/java/com/emc/object/s3/LargeFileUploaderTest.java
@@ -52,6 +52,7 @@
 import java.security.DigestInputStream;
 import java.security.MessageDigest;
 import java.util.*;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
@@ -168,6 +169,88 @@ public void testLargeFileUploaderProgressListener() throws Exception {
         client.deleteObject(getTestBucket(), key);
     }
 
+    @Test
+    public void testLFUCopy() {
+        String srcKey = "lfu-source.bin";
+        String dstKey = "lfu-single-copy.bin";
+        String dstKey2 = "lfu-multipart-copy.bin";
+
+        byte[] data = new byte[(int) MockMultipartSource.totalSize];
+        new Random().nextBytes(data);
+        client.putObject(getTestBucket(), srcKey, data, null);
+        S3ObjectMetadata sourceMetadata = client.getObjectMetadata(getTestBucket(), srcKey);
+        Assert.assertEquals(MockMultipartSource.totalSize, sourceMetadata.getContentLength().longValue());
+
+        //Single Copy
+        S3ObjectMetadata objectMetadata = new S3ObjectMetadata().addUserMetadata("key", "value");
+        LargeFileUploader uploader = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withObjectMetadata(objectMetadata);
+        uploader.upload();
+        Assert.assertEquals(0, uploader.getBytesTransferred());
+        Assert.assertEquals(sourceMetadata.getETag(), uploader.getETag());
+        Assert.assertEquals(MockMultipartSource.totalSize, client.getObjectMetadata(getTestBucket(), dstKey).getContentLength().longValue());
+        Assert.assertEquals(objectMetadata.getUserMetadata(), client.getObjectMetadata(getTestBucket(), dstKey).getUserMetadata());
+
+        //Multi Part Copy
+        LargeFileUploader uploader2 = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey2)
+                .withObjectMetadata(objectMetadata)
+                .withPartSize(MockMultipartSource.partSize)
+                .withMpuThreshold(MockMultipartSource.partSize);
+        uploader2.upload();
+        Assert.assertEquals(0, uploader2.getBytesTransferred());
+        Assert.assertTrue(uploader2.getETag().contains("-")); // hyphen signifies multipart / updated object
+        Assert.assertEquals(MockMultipartSource.totalSize, client.getObjectMetadata(getTestBucket(), dstKey2).getContentLength().longValue());
+        Assert.assertEquals(objectMetadata.getUserMetadata(), client.getObjectMetadata(getTestBucket(), dstKey2).getUserMetadata());
+    }
+
+    @Test
+    public void testLFUCopyPauseResume() throws Exception {
+        String srcKey = "mpu-copy-pause.source";
+        String dstKey = "mpu-copy-pause.target";
+
+        MockMultipartSource mockMultipartSource = new MockMultipartSource();
+        LargeFileUploader lfu = new TestLargeFileUploader(client, getTestBucket(), srcKey, mockMultipartSource)
+                .withPartSize(mockMultipartSource.getPartSize()).withMpuThreshold((int) mockMultipartSource.getTotalSize());
+        lfu.doMultipartUpload();
+
+        lfu = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withPartSize(mockMultipartSource.getPartSize()).withMpuThreshold((int) mockMultipartSource.getTotalSize())
+                .withThreads(1);
+        LargeFileUpload upload = lfu.uploadAsync();
+
+        // wait for first a few parts to start
+        while (lfu.getExecutorService() == null || ((ThreadPoolExecutor) lfu.getExecutorService()).getCompletedTaskCount() == 0) {
+            Thread.sleep(100);
+        }
+        LargeFileUploaderResumeContext resumeContext = upload.pause();
+
+        // object should not exist
+        try {
+            Assert.assertNull(client.getObjectMetadata(getTestBucket(), dstKey));
+        } catch (S3Exception e) {
+            Assert.assertEquals(404, e.getHttpCode());
+            Assert.assertEquals("NoSuchKey", e.getErrorCode());
+        }
+
+        // check resume context accuracy
+        Assert.assertNotNull(resumeContext.getUploadId());
+        Assert.assertNotNull(resumeContext.getUploadedParts());
+        Assert.assertFalse(resumeContext.getUploadedParts().isEmpty());
+        List<MultipartPart> parts = client.listParts(getTestBucket(), dstKey, resumeContext.getUploadId()).getParts();
+        Assert.assertNotNull(parts);
+
+        lfu = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withPartSize(mockMultipartSource.getPartSize()).withMpuThreshold((int) mockMultipartSource.getTotalSize())
+                .withResumeContext(resumeContext);
+        lfu.doMultipartUpload();
+
+        // Unfortunately, MPU Etag is not preserved even if copy parts matches the original multipart upload.
+        // So the content verification is done by calculating md5sum instead of checking ETag.
+        //Assert.assertEquals(mockMultipartSource.getMpuETag(), client.getObjectMetadata(getTestBucket(), dstKey).getETag());
+        Assert.assertEquals(DigestUtils.md5Hex(client.getObject(getTestBucket(), srcKey).getObject()), DigestUtils.md5Hex(client.getObject(getTestBucket(), dstKey).getObject()));
+        Assert.assertEquals(mockMultipartSource.getTotalSize(), client.getObjectMetadata(getTestBucket(), dstKey).getContentLength().longValue());
+    }
+
     @Test
     public void testLargeFileUploaderStream() {
         String key = "large-file-uploader-stream.bin";
diff --git a/src/test/java/com/emc/object/s3/TestLargeFileUploader.java b/src/test/java/com/emc/object/s3/TestLargeFileUploader.java
@@ -45,6 +45,9 @@ public TestLargeFileUploader(S3Client s3Client, String bucket, String key, Large
         super(s3Client, bucket, key, multipartSource);
     }
 
+    public TestLargeFileUploader(S3Client s3Client, String srcBucket, String srcKey, String dstBucket, String dstKey){
+        super(s3Client, srcBucket, srcKey, dstBucket, dstKey);
+    }
     @Override
     protected long getMinPartSize() {
         return 100 * 1024; // 100 KiB for testing

Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,9 @@ public TestLargeFileUploader(S3Client s3Client, String bucket, String key, Large`
`45`	`45`	`super(s3Client, bucket, key, multipartSource);`
`46`	`46`	`}`
`47`	`47`
	`48`	`+ public TestLargeFileUploader(S3Client s3Client, String srcBucket, String srcKey, String dstBucket, String dstKey){`
	`49`	`+ super(s3Client, srcBucket, srcKey, dstBucket, dstKey);`
	`50`	`+ }`
`48`	`51`	`@Override`
`49`	`52`	`protected long getMinPartSize() {`
`50`	`53`	`return 100 * 1024; // 100 KiB for testing`