Upgrade GetObject to use SDK v2.

Key change: `getObject` now returns a `ResponseInputStream<GetObjectResponse>` rather than a `S3Object`. This makes it simpler to handle the input stream lifetime in various classes such as `S3AInputStream`, `S3ARemoteObject`, or `SDKStreamDrainer`.

Author: passaro

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

@@ -59,13 +59,11 @@
 import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsResult;
-import com.amazonaws.services.s3.model.GetObjectRequest;
 import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
 import com.amazonaws.services.s3.model.InitiateMultipartUploadResult;
 import com.amazonaws.services.s3.model.ListMultipartUploadsRequest;
 import com.amazonaws.services.s3.model.MultiObjectDeleteException;
 import com.amazonaws.services.s3.model.MultipartUpload;
-import com.amazonaws.services.s3.model.S3Object;
 import com.amazonaws.services.s3.model.SelectObjectContentRequest;
 import com.amazonaws.services.s3.model.SelectObjectContentResult;
 import com.amazonaws.services.s3.model.StorageClass;
@@ -78,7 +76,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
 import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
 import software.amazon.awssdk.awscore.exception.AwsServiceException;
@@ -1643,18 +1644,18 @@ public void close()  {
     }
 
     @Override
-    public GetObjectRequest newGetRequest(final String key) {
+    public GetObjectRequest.Builder newGetRequestBuilder(final String key) {
       // active the audit span used for the operation
       try (AuditSpan span = auditSpan.activate()) {
-        return getRequestFactory().newGetObjectRequest(key);
+        return getRequestFactory().newGetObjectRequestBuilder(key);
       }
     }
 
     @Override
-    public S3Object getObject(GetObjectRequest request) {
+    public ResponseInputStream<GetObjectResponse> getObject(GetObjectRequest request) {
       // active the audit span used for the operation
       try (AuditSpan span = auditSpan.activate()) {
-        return s3.getObject(request);
+        return s3V2.getObject(request);
       }
     }
 

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

@@ -22,6 +22,7 @@
 import java.io.Closeable;
 import java.io.EOFException;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.InterruptedIOException;
 import java.net.SocketTimeoutException;
 import java.nio.ByteBuffer;
@@ -31,9 +32,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.IntFunction;
 
-import com.amazonaws.services.s3.model.GetObjectRequest;
-import com.amazonaws.services.s3.model.S3Object;
-import com.amazonaws.services.s3.model.S3ObjectInputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,7 +44,6 @@
 import org.apache.hadoop.fs.FSExceptionMessages;
 import org.apache.hadoop.fs.FSInputStream;
 import org.apache.hadoop.fs.FileRange;
-import org.apache.hadoop.fs.PathIOException;
 import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.fs.impl.CombinedFileRange;
 import org.apache.hadoop.fs.VectoredReadUtils;
@@ -61,6 +58,10 @@
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.util.functional.CallableRaisingIOE;
 
+import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+
 import static java.util.Objects.requireNonNull;
 import static org.apache.commons.lang3.StringUtils.isNotEmpty;
 import static org.apache.hadoop.fs.VectoredReadUtils.isOrderedDisjoint;
@@ -125,14 +126,9 @@ public class S3AInputStream extends FSInputStream implements  CanSetReadahead,
    */
   private volatile boolean closed;
   /**
-   * wrappedStream is associated with an object (instance of S3Object). When
-   * the object is garbage collected, the associated wrappedStream will be
-   * closed. Keep a reference to this object to prevent the wrapperStream
-   * still in use from being closed unexpectedly due to garbage collection.
-   * See HADOOP-17338 for details.
+   * Input stream returned by a getObject call.
    */
-  private S3Object object;
-  private S3ObjectInputStream wrappedStream;
+  private ResponseInputStream<GetObjectResponse> wrappedStream;
   private final S3AReadOpContext context;
   private final InputStreamCallbacks client;
 
@@ -271,28 +267,22 @@ private synchronized void reopen(String reason, long targetPos, long length,
         uri, reason, targetPos, contentRangeFinish, length,  pos, nextReadPos,
         inputPolicy);
 
+    GetObjectRequest request = client.newGetRequestBuilder(key)
+        .range(S3AUtils.formatRange(targetPos, contentRangeFinish - 1))
+        .applyMutation(changeTracker::maybeApplyConstraint)
+        .build();
     long opencount = streamStatistics.streamOpened();
-    GetObjectRequest request = client.newGetRequest(key)
-        .withRange(targetPos, contentRangeFinish - 1);
     String operation = opencount == 0 ? OPERATION_OPEN : OPERATION_REOPEN;
     String text = String.format("%s %s at %d",
         operation, uri, targetPos);
-    changeTracker.maybeApplyConstraint(request);
-
-    object = onceTrackingDuration(text, uri,
+    wrappedStream = onceTrackingDuration(text, uri,
         streamStatistics.initiateGetRequest(), () ->
             client.getObject(request));
 
-
-    changeTracker.processResponse(object, operation,
+    changeTracker.processResponse(wrappedStream.response(), operation,
         targetPos);
-    wrappedStream = object.getObjectContent();
-    contentRangeStart = targetPos;
-    if (wrappedStream == null) {
-      throw new PathIOException(uri,
-          "Null IO stream from " + operation + " of (" + reason +  ") ");
-    }
 
+    contentRangeStart = targetPos;
     this.pos = targetPos;
   }
 
@@ -505,14 +495,15 @@ public synchronized int read() throws IOException {
    */
   @Retries.OnceTranslated
   private void onReadFailure(IOException ioe, boolean forceAbort) {
+    GetObjectResponse objectResponse = wrappedStream == null ? null : wrappedStream.response();
     if (LOG.isDebugEnabled()) {
       LOG.debug("Got exception while trying to read from stream {}, " +
           "client: {} object: {}, trying to recover: ",
-          uri, client, object, ioe);
+          uri, client, objectResponse, ioe);
     } else {
       LOG.info("Got exception while trying to read from stream {}, " +
           "client: {} object: {}, trying to recover: " + ioe,
-          uri, client, object);
+          uri, client, objectResponse);
     }
     streamStatistics.readException();
     closeStream("failure recovery", forceAbort, false);
@@ -672,7 +663,6 @@ private CompletableFuture<Boolean> closeStream(
     CompletableFuture<Boolean> operation;
     SDKStreamDrainer drainer = new SDKStreamDrainer(
         uri,
-        object,
         wrappedStream,
         shouldAbort,
         (int) remaining,
@@ -694,7 +684,6 @@ private CompletableFuture<Boolean> closeStream(
     // either the stream is closed in the blocking call or the async call is
     // submitted with its own copy of the references
     wrappedStream = null;
-    object = null;
     return operation;
   }
 
@@ -910,29 +899,21 @@ public void readVectored(List<? extends FileRange> ranges,
   private void readCombinedRangeAndUpdateChildren(CombinedFileRange combinedFileRange,
                                                   IntFunction<ByteBuffer> allocate) {
     LOG.debug("Start reading combined range {} from path {} ", combinedFileRange, pathStr);
-    // This reference is must be kept till all buffers are populated as this is a
-    // finalizable object which closes the internal stream when gc triggers.
-    S3Object objectRange = null;
-    S3ObjectInputStream objectContent = null;
+    ResponseInputStream<GetObjectResponse> rangeContent = null;
     try {
       checkIfVectoredIOStopped();
       final String operationName = "readCombinedFileRange";
-      objectRange = getS3Object(operationName,
+      rangeContent = getS3Object(operationName,
               combinedFileRange.getOffset(),
               combinedFileRange.getLength());
-      objectContent = objectRange.getObjectContent();
-      if (objectContent == null) {
-        throw new PathIOException(uri,
-                "Null IO stream received during " + operationName);
-      }
-      populateChildBuffers(combinedFileRange, objectContent, allocate);
+      populateChildBuffers(combinedFileRange, rangeContent, allocate);
     } catch (Exception ex) {
       LOG.debug("Exception while reading a range {} from path {} ", combinedFileRange, pathStr, ex);
       for(FileRange child : combinedFileRange.getUnderlying()) {
         child.getData().completeExceptionally(ex);
       }
     } finally {
-      IOUtils.cleanupWithLogger(LOG, objectRange, objectContent);
+      IOUtils.cleanupWithLogger(LOG, rangeContent);
     }
     LOG.debug("Finished reading range {} from path {} ", combinedFileRange, pathStr);
   }
@@ -945,7 +926,7 @@ private void readCombinedRangeAndUpdateChildren(CombinedFileRange combinedFileRa
    * @throws IOException any IOE.
    */
   private void populateChildBuffers(CombinedFileRange combinedFileRange,
-                                    S3ObjectInputStream objectContent,
+                                    InputStream objectContent,
                                     IntFunction<ByteBuffer> allocate) throws IOException {
     // If the combined file range just contains a single child
     // range, we only have to fill that one child buffer else
@@ -977,7 +958,7 @@ private void populateChildBuffers(CombinedFileRange combinedFileRange,
    * @param drainQuantity how many bytes to drain.
    * @throws IOException any IOE.
    */
-  private void drainUnnecessaryData(S3ObjectInputStream objectContent, long drainQuantity)
+  private void drainUnnecessaryData(InputStream objectContent, long drainQuantity)
           throws IOException {
     int drainBytes = 0;
     int readCount;
@@ -1019,26 +1000,20 @@ private void validateRangeRequest(FileRange range) throws EOFException {
    */
   private void readSingleRange(FileRange range, ByteBuffer buffer) {
     LOG.debug("Start reading range {} from path {} ", range, pathStr);
-    S3Object objectRange = null;
-    S3ObjectInputStream objectContent = null;
+    ResponseInputStream<GetObjectResponse> objectRange = null;
     try {
       checkIfVectoredIOStopped();
       long position = range.getOffset();
       int length = range.getLength();
       final String operationName = "readRange";
       objectRange = getS3Object(operationName, position, length);
-      objectContent = objectRange.getObjectContent();
-      if (objectContent == null) {
-        throw new PathIOException(uri,
-                "Null IO stream received during " + operationName);
-      }
-      populateBuffer(length, buffer, objectContent);
+      populateBuffer(length, buffer, objectRange);
       range.getData().complete(buffer);
     } catch (Exception ex) {
       LOG.warn("Exception while reading a range {} from path {} ", range, pathStr, ex);
       range.getData().completeExceptionally(ex);
     } finally {
-      IOUtils.cleanupWithLogger(LOG, objectRange, objectContent);
+      IOUtils.cleanupWithLogger(LOG, objectRange);
     }
     LOG.debug("Finished reading range {} from path {} ", range, pathStr);
   }
@@ -1053,7 +1028,7 @@ private void readSingleRange(FileRange range, ByteBuffer buffer) {
    */
   private void populateBuffer(int length,
                               ByteBuffer buffer,
-                              S3ObjectInputStream objectContent) throws IOException {
+                              InputStream objectContent) throws IOException {
 
     if (buffer.isDirect()) {
       VectoredReadUtils.readInDirectBuffer(length, buffer,
@@ -1078,7 +1053,7 @@ private void populateBuffer(int length,
    * @param length number of bytes to fill in dest.
    * @throws IOException any IOE.
    */
-  private void readByteArray(S3ObjectInputStream objectContent,
+  private void readByteArray(InputStream objectContent,
                             byte[] dest,
                             int offset,
                             int length) throws IOException {
@@ -1105,13 +1080,16 @@ private void readByteArray(S3ObjectInputStream objectContent,
    * @return S3Object result s3 object.
    * @throws IOException exception if any.
    */
-  private S3Object getS3Object(String operationName, long position,
-                               int length) throws IOException {
-    final GetObjectRequest request = client.newGetRequest(key)
-            .withRange(position, position + length - 1);
-    changeTracker.maybeApplyConstraint(request);
+  private ResponseInputStream<GetObjectResponse> getS3Object(String operationName,
+                                                             long position,
+                                                             int length)
+      throws IOException {
+    final GetObjectRequest request = client.newGetRequestBuilder(key)
+        .range(S3AUtils.formatRange(position, position + length - 1))
+        .applyMutation(changeTracker::maybeApplyConstraint)
+        .build();
     DurationTracker tracker = streamStatistics.initiateGetRequest();
-    S3Object objectRange;
+    ResponseInputStream<GetObjectResponse> objectRange;
     Invoker invoker = context.getReadInvoker();
     try {
       objectRange = invoker.retry(operationName, pathStr, true,
@@ -1126,7 +1104,7 @@ private S3Object getS3Object(String operationName, long position,
     } finally {
       tracker.close();
     }
-    changeTracker.processResponse(objectRange, operationName,
+    changeTracker.processResponse(objectRange.response(), operationName,
             position);
     return objectRange;
   }
@@ -1279,19 +1257,19 @@ public IOStatistics getIOStatistics() {
   public interface InputStreamCallbacks extends Closeable {
 
     /**
-     * Create a GET request.
+     * Create a GET request builder.
      * @param key object key
-     * @return the request
+     * @return the request builder
      */
-    GetObjectRequest newGetRequest(String key);
+    GetObjectRequest.Builder newGetRequestBuilder(String key);
 
     /**
      * Execute the request.
      * @param request the request
      * @return the response
      */
     @Retries.OnceRaw
-    S3Object getObject(GetObjectRequest request);
+    ResponseInputStream<GetObjectResponse> getObject(GetObjectRequest request);
 
     /**
      * Submit some asynchronous work, for example, draining a stream.

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

@@ -1932,4 +1932,15 @@ public String toString() {
     }
   };
 
+  /**
+   * Format a byte range for a request header.
+   * See https://www.rfc-editor.org/rfc/rfc9110.html#section-14.1.2
+   *
+   * @param rangeStart the start byte offset
+   * @param rangeEnd the end byte offset (inclusive)
+   * @return a formatted byte range
+   */
+  public static String formatRange(long rangeStart, long rangeEnd) {
+    return String.format("bytes={}-{}", rangeStart, rangeEnd);
+  }
 }

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

@@ -29,30 +29,30 @@
 import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
 import com.amazonaws.services.s3.model.DeleteObjectRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
-import com.amazonaws.services.s3.model.GetObjectRequest;
 import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
 import com.amazonaws.services.s3.model.ListMultipartUploadsRequest;
 import com.amazonaws.services.s3.model.ListNextBatchOfObjectsRequest;
-
-import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
-import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
-import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
-import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
 import com.amazonaws.services.s3.model.ObjectListing;
 import com.amazonaws.services.s3.model.PartETag;
 import com.amazonaws.services.s3.model.SSEAwsKeyManagementParams;
 import com.amazonaws.services.s3.model.SSECustomerKey;
 import com.amazonaws.services.s3.model.SelectObjectContentRequest;
 import com.amazonaws.services.s3.model.StorageClass;
 import com.amazonaws.services.s3.model.UploadPartRequest;
-import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 
 import org.apache.hadoop.fs.PathIOException;
 import org.apache.hadoop.fs.s3a.S3AEncryptionMethods;
 import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecrets;
 import org.apache.hadoop.fs.s3a.impl.PutObjectOptions;
 
+import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+
 /**
  * Factory for S3 objects.
  *
@@ -202,11 +202,11 @@ CompleteMultipartUploadRequest newCompleteMultipartUploadRequest(
 
 
   /**
-   * Create a GET request.
+   * Create a GET request builder.
    * @param key object key
-   * @return the request.
+   * @return the request builder.
    */
-  GetObjectRequest newGetObjectRequest(String key);
+  GetObjectRequest.Builder newGetObjectRequestBuilder(String key);
 
   /**
    * Create and initialize a part request of a multipart upload.