HADOOP-15667. FileSystemMultipartUploader should verify that UploadHandle has non-0 length.

Contributed by Ewan Higgs

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystemMultipartUploader.java

@@ -68,6 +68,7 @@ public class FileSystemMultipartUploader extends MultipartUploader {
       throws IOException {
 
     byte[] uploadIdByteArray = uploadId.toByteArray();
+    checkUploadId(uploadIdByteArray);
     Path collectorPath = new Path(new String(uploadIdByteArray, 0,
         uploadIdByteArray.length, Charsets.UTF_8));
     Path partPath =
@@ -101,6 +102,8 @@ public class FileSystemMultipartUploader extends MultipartUploader {
       List<Pair<Integer, PartHandle>> handles, UploadHandle multipartUploadId)
       throws IOException {
 
+    checkUploadId(multipartUploadId.toByteArray());
+
     if (handles.isEmpty()) {
       throw new IOException("Empty upload");
     }
@@ -133,8 +136,7 @@ public class FileSystemMultipartUploader extends MultipartUploader {
   @Override
   public void abort(Path filePath, UploadHandle uploadId) throws IOException {
     byte[] uploadIdByteArray = uploadId.toByteArray();
-    Preconditions.checkArgument(uploadIdByteArray.length != 0,
-        "UploadId is empty");
+    checkUploadId(uploadIdByteArray);
     Path collectorPath = new Path(new String(uploadIdByteArray, 0,
         uploadIdByteArray.length, Charsets.UTF_8));
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/MultipartUploader.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.util.List;
 
+import com.google.common.base.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -91,4 +92,14 @@ public abstract class MultipartUploader {
   public abstract void abort(Path filePath, UploadHandle multipartUploadId)
       throws IOException;
 
+  /**
+   * Utility method to validate uploadIDs
+   * @param uploadId
+   * @throws IllegalArgumentException
+   */
+  protected void checkUploadId(byte[] uploadId)
+      throws IllegalArgumentException {
+    Preconditions.checkArgument(uploadId.length > 0,
+        "Empty UploadId is not valid");
+  }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractMultipartUploaderTest.java

@@ -297,4 +297,47 @@ public abstract class AbstractContractMultipartUploaderTest extends
         () -> mpu.complete(dest, new ArrayList<>(), handle));
     mpu.abort(dest, handle);
   }
+
+  /**
+   * When we pass empty uploadID, putPart throws IllegalArgumentException.
+   * @throws Exception
+   */
+  @Test
+  public void testPutPartEmptyUploadID() throws Exception {
+    describe("Expect IllegalArgumentException when putPart uploadID is empty");
+    FileSystem fs = getFileSystem();
+    Path dest = path("testCompleteEmptyUpload");
+    MultipartUploader mpu = MultipartUploaderFactory.get(fs, null);
+    mpu.initialize(dest);
+    UploadHandle emptyHandle =
+        BBUploadHandle.from(ByteBuffer.wrap(new byte[0]));
+    byte[] payload = generatePayload(1);
+    InputStream is = new ByteArrayInputStream(payload);
+    intercept(IllegalArgumentException.class,
+        () -> mpu.putPart(dest, is, 1, emptyHandle, payload.length));
+  }
+
+  /**
+   * When we pass empty uploadID, complete throws IllegalArgumentException.
+   * @throws Exception
+   */
+  @Test
+  public void testCompleteEmptyUploadID() throws Exception {
+    describe("Expect IllegalArgumentException when complete uploadID is empty");
+    FileSystem fs = getFileSystem();
+    Path dest = path("testCompleteEmptyUpload");
+    MultipartUploader mpu = MultipartUploaderFactory.get(fs, null);
+    UploadHandle realHandle = mpu.initialize(dest);
+    UploadHandle emptyHandle =
+        BBUploadHandle.from(ByteBuffer.wrap(new byte[0]));
+    List<Pair<Integer, PartHandle>> partHandles = new ArrayList<>();
+    byte[] payload = generatePayload(1);
+    InputStream is = new ByteArrayInputStream(payload);
+    PartHandle partHandle = mpu.putPart(dest, is, 1, realHandle,
+        payload.length);
+    partHandles.add(Pair.of(1, partHandle));
+
+    intercept(IllegalArgumentException.class,
+        () -> mpu.complete(dest, partHandles, emptyHandle));
+  }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AMultipartUploader.java

@@ -84,9 +84,10 @@ public class S3AMultipartUploader extends MultipartUploader {
   public PartHandle putPart(Path filePath, InputStream inputStream,
       int partNumber, UploadHandle uploadId, long lengthInBytes)
       throws IOException {
-    final WriteOperationHelper writeHelper = s3a.getWriteOperationHelper();
-    String key = s3a.pathToKey(filePath);
     byte[] uploadIdBytes = uploadId.toByteArray();
+    checkUploadId(uploadIdBytes);
+    String key = s3a.pathToKey(filePath);
+    final WriteOperationHelper writeHelper = s3a.getWriteOperationHelper();
     String uploadIdString = new String(uploadIdBytes, 0, uploadIdBytes.length,
         Charsets.UTF_8);
     UploadPartRequest request = writeHelper.newUploadPartRequest(key,
@@ -155,11 +156,6 @@ public class S3AMultipartUploader extends MultipartUploader {
     }
   }
 
-  private void checkUploadId(byte[] uploadId) throws IllegalArgumentException {
-    Preconditions.checkArgument(uploadId.length > 0,
-        "Empty UploadId is not valid");
-  }
-
   /**
    * Build the payload for marshalling.
    * @param eTag upload etag