Address PR Comments

Author: shameersss1

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

@@ -57,6 +57,7 @@ CompletableFuture<UploadHandle> startUpload(Path filePath)
    * It is possible to have parts uploaded in any order (or in parallel).
    * @param uploadId Identifier from {@link #startUpload(Path)}.
    * @param partNumber Index of the part relative to others.
+   * @param isLastPart  is the part the last part of the upload?
    * @param filePath Target path for upload (as {@link #startUpload(Path)}).
    * @param inputStream Data for this part. Implementations MUST close this
    * stream after reading in the data.
@@ -67,6 +68,7 @@ CompletableFuture<UploadHandle> startUpload(Path filePath)
   CompletableFuture<PartHandle> putPart(
       UploadHandle uploadId,
       int partNumber,
+      boolean isLastPart,
       Path filePath,
       InputStream inputStream,
       long lengthInBytes)
@@ -77,7 +79,7 @@ CompletableFuture<PartHandle> putPart(
    * @param uploadId Identifier from {@link #startUpload(Path)}.
    * @param filePath Target path for upload (as {@link #startUpload(Path)}.
    * @param handles non-empty map of part number to part handle.
-   *          from {@link #putPart(UploadHandle, int, Path, InputStream, long)}.
+   *          from {@link #putPart(UploadHandle, int, boolean, Path, InputStream, long)}.
    * @return unique PathHandle identifier for the uploaded file.
    * @throws IOException IO failure
    */

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

@@ -101,7 +101,7 @@ protected void checkPartHandles(Map<Integer, PartHandle> partHandles) {
 
   /**
    * Check all the arguments to the
-   * {@link MultipartUploader#putPart(UploadHandle, int, Path, InputStream, long)}
+   * {@link MultipartUploader#putPart(UploadHandle, int, boolean, Path, InputStream, long)}
    * operation.
    * @param filePath Target path for upload (as {@link #startUpload(Path)}).
    * @param inputStream Data for this part. Implementations MUST close this

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

@@ -111,7 +111,7 @@ public CompletableFuture<UploadHandle> startUpload(Path filePath)
 
   @Override
   public CompletableFuture<PartHandle> putPart(UploadHandle uploadId,
-      int partNumber, Path filePath,
+      int partNumber, boolean isLastPart, Path filePath,
       InputStream inputStream,
       long lengthInBytes)
       throws IOException {

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

@@ -249,7 +249,7 @@ public void testSingleUpload() throws Exception {
     // was interpreted as an inconsistent write.
     MultipartUploader completer = uploader0;
     // and upload with uploader 1 to validate cross-uploader uploads
-    PartHandle partHandle = putPart(file, uploadHandle, 1, payload);
+    PartHandle partHandle = putPart(file, uploadHandle, 1, true, payload);
     partHandles.put(1, partHandle);
     PathHandle fd = complete(completer, uploadHandle, file,
         partHandles);
@@ -317,12 +317,13 @@ protected PartHandle buildAndPutPart(
       final Path file,
       final UploadHandle uploadHandle,
       final int index,
+      final boolean isLastPart,
       final MessageDigest origDigest) throws IOException {
     byte[] payload = generatePayload(index);
     if (origDigest != null) {
       origDigest.update(payload);
     }
-    return putPart(file, uploadHandle, index, payload);
+    return putPart(file, uploadHandle, index, isLastPart, payload);
   }
 
   /**
@@ -331,13 +332,15 @@ protected PartHandle buildAndPutPart(
    * @param file destination
    * @param uploadHandle handle
    * @param index index of part
+   * @param isLastPart is last part of the upload ?
    * @param payload byte array of payload
    * @return the part handle
    * @throws IOException IO failure.
    */
   protected PartHandle putPart(final Path file,
       final UploadHandle uploadHandle,
       final int index,
+      final boolean isLastPart,
       final byte[] payload) throws IOException {
     ContractTestUtils.NanoTimer timer = new ContractTestUtils.NanoTimer();
     PartHandle partHandle;
@@ -347,7 +350,7 @@ protected PartHandle putPart(final Path file,
                  payload.length,
                  file)) {
       partHandle = awaitFuture(getUploader(index)
-          .putPart(uploadHandle, index, file,
+          .putPart(uploadHandle, index, isLastPart, file,
               new ByteArrayInputStream(payload),
               payload.length));
     }
@@ -488,7 +491,7 @@ public void testMultipartUpload() throws Exception {
     MessageDigest origDigest = DigestUtils.getMd5Digest();
     int payloadCount = getTestPayloadCount();
     for (int i = 1; i <= payloadCount; ++i) {
-      PartHandle partHandle = buildAndPutPart(file, uploadHandle, i,
+      PartHandle partHandle = buildAndPutPart(file, uploadHandle, i, i == payloadCount,
           origDigest);
       partHandles.put(i, partHandle);
     }
@@ -515,7 +518,7 @@ public void testMultipartUploadEmptyPart() throws Exception {
       origDigest.update(payload);
       InputStream is = new ByteArrayInputStream(payload);
       PartHandle partHandle = awaitFuture(
-          uploader.putPart(uploadHandle, 1, file, is, payload.length));
+          uploader.putPart(uploadHandle, 1, true, file, is, payload.length));
       partHandles.put(1, partHandle);
       completeUpload(file, uploadHandle, partHandles, origDigest, 0);
     }
@@ -530,7 +533,7 @@ public void testUploadEmptyBlock() throws Exception {
     Path file = methodPath();
     UploadHandle uploadHandle = startUpload(file);
     Map<Integer, PartHandle> partHandles = new HashMap<>();
-    partHandles.put(1, putPart(file, uploadHandle, 1, new byte[0]));
+    partHandles.put(1, putPart(file, uploadHandle, 1, true, new byte[0]));
     completeUpload(file, uploadHandle, partHandles, null, 0);
   }
 
@@ -550,7 +553,8 @@ public void testMultipartUploadReverseOrder() throws Exception {
       origDigest.update(payload);
     }
     for (int i = payloadCount; i > 0; --i) {
-      partHandles.put(i, buildAndPutPart(file, uploadHandle, i, null));
+      partHandles.put(i, buildAndPutPart(file, uploadHandle, i, i == payloadCount,
+          null));
     }
     completeUpload(file, uploadHandle, partHandles, origDigest,
         payloadCount * partSizeInBytes());
@@ -574,7 +578,8 @@ public void testMultipartUploadReverseOrderNonContiguousPartNumbers()
     }
     Map<Integer, PartHandle> partHandles = new HashMap<>();
     for (int i = payloadCount; i > 0; i -= 2) {
-      partHandles.put(i, buildAndPutPart(file, uploadHandle, i, null));
+      partHandles.put(i, buildAndPutPart(file, uploadHandle, i, i == payloadCount,
+          null));
     }
     completeUpload(file, uploadHandle, partHandles, origDigest,
         getTestPayloadCount() * partSizeInBytes());
@@ -591,7 +596,7 @@ public void testMultipartUploadAbort() throws Exception {
     UploadHandle uploadHandle = startUpload(file);
     Map<Integer, PartHandle> partHandles = new HashMap<>();
     for (int i = 12; i > 10; i--) {
-      partHandles.put(i, buildAndPutPart(file, uploadHandle, i, null));
+      partHandles.put(i, buildAndPutPart(file, uploadHandle, i, i == 12, null));
     }
     abortUpload(uploadHandle, file);
 
@@ -601,7 +606,7 @@ public void testMultipartUploadAbort() throws Exception {
 
     intercept(IOException.class,
         () -> awaitFuture(
-            uploader0.putPart(uploadHandle, 49, file, is, len)));
+            uploader0.putPart(uploadHandle, 49, true, file, is, len)));
     intercept(IOException.class,
         () -> complete(uploader0, uploadHandle, file, partHandles));
 
@@ -701,7 +706,8 @@ public void testPutPartEmptyUploadID() throws Exception {
     byte[] payload = generatePayload(1);
     InputStream is = new ByteArrayInputStream(payload);
     intercept(IllegalArgumentException.class,
-        () -> uploader0.putPart(emptyHandle, 1, dest, is, payload.length));
+        () -> uploader0.putPart(emptyHandle, 1, true, dest, is,
+            payload.length));
   }
 
   /**
@@ -715,7 +721,7 @@ public void testCompleteEmptyUploadID() throws Exception {
     UploadHandle emptyHandle =
         BBUploadHandle.from(ByteBuffer.wrap(new byte[0]));
     Map<Integer, PartHandle> partHandles = new HashMap<>();
-    PartHandle partHandle = putPart(dest, realHandle, 1,
+    PartHandle partHandle = putPart(dest, realHandle, 1, true,
         generatePayload(1, SMALL_FILE));
     partHandles.put(1, partHandle);
 
@@ -743,7 +749,7 @@ public void testDirectoryInTheWay() throws Exception {
     UploadHandle uploadHandle = startUpload(file);
     Map<Integer, PartHandle> partHandles = new HashMap<>();
     int size = SMALL_FILE;
-    PartHandle partHandle = putPart(file, uploadHandle, 1,
+    PartHandle partHandle = putPart(file, uploadHandle, 1, true,
         generatePayload(1, size));
     partHandles.put(1, partHandle);
 
@@ -802,10 +808,10 @@ public void testConcurrentUploads() throws Throwable {
     assertNotEquals("Upload handles match", upload1, upload2);
 
     // put part 1
-    partHandles1.put(partId1, putPart(file, upload1, partId1, payload1));
+    partHandles1.put(partId1, putPart(file, upload1, partId1, false, payload1));
 
     // put part2
-    partHandles2.put(partId2, putPart(file, upload2, partId2, payload2));
+    partHandles2.put(partId2, putPart(file, upload2, partId2, true, payload2));
 
     // complete part u1. expect its size and digest to
     // be as expected.

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

@@ -737,14 +737,16 @@ private Constants() {
       "fs.s3a.encryption.key";
 
   /**
-   * Client side encryption (CSE-CUSTOM) with custom cryptographic material manager class name.
+   * Custom keyring class name for CSE-KMS.
+   * value:{@value}
    */
   public static final String S3_ENCRYPTION_CSE_CUSTOM_KEYRING_CLASS_NAME =
           "fs.s3a.encryption.cse.custom.keyring.class.name";
 
   /**
-   * This config initializes unencrypted s3 client will be used to access unencrypted
-   * s3 object. This is to provide backward compatibility.
+   * Config to support reading unencrypted s3 objects when CSE is enabled.
+   * This is to provide backward compatibility with V1/V2 client.
+   * value:{@value}
    */
   public static final String S3_ENCRYPTION_CSE_READ_UNENCRYPTED_OBJECTS =
           "fs.s3a.encryption.cse.read.unencrypted.objects";
@@ -759,7 +761,8 @@ private Constants() {
    * This is to provide backward compatability with objects encrypted with V1 client.
    * Unlike V2 and V3 client which always pads 16 bytes, V1 client pads bytes till the
    * object size reaches next multiple of 16.
-   * * This is to provide backward compatibility.
+   * This is to provide backward compatibility with V1 client.
+   * value:{@value}
    */
   public static final String S3_ENCRYPTION_CSE_OBJECT_SIZE_FROM_RANGED_GET_ENABLED =
           "fs.s3a.encryption.cse.object.size.ranged.get.enabled";
@@ -770,10 +773,11 @@ private Constants() {
   public static final boolean S3_ENCRYPTION_CSE_OBJECT_SIZE_FROM_RANGED_GET_ENABLED_DEFAULT = false;
 
   /**
-   * Config to control whether to skip file named with suffix
-   * {@link #S3_ENCRYPTION_CSE_INSTRUCTION_FILE_SUFFIX}. Encryption V1 client supports storing
-   * encryption metadata in an instruction file which should be skipped while listing for the files.
-   * This is to provide backward compatibility.
+   * Config to skip file named with suffix
+   * {@link #S3_ENCRYPTION_CSE_INSTRUCTION_FILE_SUFFIX}. V1/V2 client supports storing encryption
+   * metadata in an instruction file which should be skipped while listing for the files.
+   * This is to provide backward compatibility with V1/V2 client.
+   * value:{@value}
    */
   public static final String S3_ENCRYPTION_CSE_SKIP_INSTRUCTION_FILE =
       "fs.s3a.encryption.cse.skip.instruction.file";

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

@@ -21,7 +21,12 @@
 import java.io.IOException;
 import java.net.URI;
 
-import org.apache.hadoop.thirdparty.com.google.common.base.Strings;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.impl.encryption.CSEMaterials;
+import org.apache.hadoop.util.Preconditions;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.functional.LazyAtomicReference;
+
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.encryption.s3.S3AsyncEncryptionClient;
@@ -30,12 +35,11 @@
 import software.amazon.encryption.s3.materials.DefaultCryptoMaterialsManager;
 import software.amazon.encryption.s3.materials.Keyring;
 
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.s3a.impl.CSEMaterials;
-import org.apache.hadoop.util.ReflectionUtils;
-
 import static org.apache.hadoop.fs.s3a.impl.InstantiationIOException.unavailable;
 
+/**
+ * Factory class to create encrypted s3 client and encrypted async s3 client.
+ */
 public class EncryptionS3ClientFactory extends DefaultS3ClientFactory {
 
   private static final String ENCRYPTION_CLIENT_CLASSNAME =
@@ -44,7 +48,11 @@ public class EncryptionS3ClientFactory extends DefaultS3ClientFactory {
   /**
    * Encryption client availability.
    */
-  private static final boolean ENCRYPTION_CLIENT_FOUND = checkForEncryptionClient();
+  private static final LazyAtomicReference<Boolean> ENCRYPTION_CLIENT_AVAILABLE =
+      LazyAtomicReference.lazyAtomicReferenceFromSupplier(
+          EncryptionS3ClientFactory::checkForEncryptionClient
+      );
+
 
   /**
    * S3Client to be wrapped by encryption client.
@@ -73,11 +81,13 @@ private static boolean checkForEncryptionClient() {
    * @return true if it was found in the classloader
    */
   private static synchronized boolean isEncryptionClientAvailable() {
-    return ENCRYPTION_CLIENT_FOUND;
+    return ENCRYPTION_CLIENT_AVAILABLE.get();
   }
 
   /**
-   * Create encrypted s3 client.
+   * Creates both synchronous and asynchronous encrypted s3 clients.
+   * Synchronous client is wrapped by encryption client first and then
+   * Asynchronous client is wrapped by encryption client.
    * @param uri S3A file system URI
    * @param parameters parameter object
    * @return encrypted s3 client
@@ -114,7 +124,16 @@ public S3AsyncClient createS3AsyncClient(URI uri, S3ClientCreationParameters par
     return createS3AsyncEncryptionClient(parameters.getClientSideEncryptionMaterials());
   }
 
+  /**
+   * Create encrypted s3 client.
+   * @param cseMaterials
+   * @return encrypted s3 client
+   */
   private S3Client createS3EncryptionClient(final CSEMaterials cseMaterials) {
+    Preconditions.checkArgument(s3AsyncClient !=null,
+        "S3 async client not initialized");
+    Preconditions.checkArgument(s3Client !=null,
+        "S3 client not initialized");
     S3EncryptionClient.Builder s3EncryptionClientBuilder =
         S3EncryptionClient.builder().wrappedAsyncClient(s3AsyncClient).wrappedClient(s3Client)
             // this is required for doing S3 ranged GET calls
@@ -141,7 +160,14 @@ private S3Client createS3EncryptionClient(final CSEMaterials cseMaterials) {
     return s3EncryptionClientBuilder.build();
   }
 
+  /**
+   * Create async encrypted s3 client.
+   * @param cseMaterials
+   * @return encrypted async s3 client
+   */
   private S3AsyncClient createS3AsyncEncryptionClient(final CSEMaterials cseMaterials) {
+    Preconditions.checkArgument(s3AsyncClient !=null,
+        "S3 async client not initialized");
     S3AsyncEncryptionClient.Builder s3EncryptionAsyncClientBuilder =
         S3AsyncEncryptionClient.builder().wrappedClient(s3AsyncClient)
             // this is required for doing S3 ranged GET calls
@@ -186,10 +212,8 @@ private Keyring getKeyringProvider(String className,
   }
 
   private Class<? extends Keyring> getCustomKeyringProviderClass(String className) {
-    if (Strings.isNullOrEmpty(className)) {
-      throw new IllegalArgumentException(
-          "Custom Keyring class name is null or empty");
-    }
+    Preconditions.checkArgument(className !=null && !className.isEmpty(),
+        "Custom Keyring class name is null or empty");
     try {
       return Class.forName(className).asSubclass(Keyring.class);
     } catch (ClassNotFoundException e) {