diff --git a/.idea/codeStyles/Project.xml b/.idea/codeStyles/Project.xml
index 82511b2671f7..31e3dd29a01f 100644
--- a/.idea/codeStyles/Project.xml
+++ b/.idea/codeStyles/Project.xml
@@ -10,6 +10,7 @@
       </option>
       <option name="IMPORT_LAYOUT_TABLE">
         <value>
+          <package name="" withSubpackages="true" static="false" module="true" />
           <package name="" withSubpackages="true" static="true" />
           <emptyLine />
           <package name="" withSubpackages="true" static="false" />
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java
index d7c872d89289..7db9a2a740c4 100644
--- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java
+++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java
@@ -348,6 +348,15 @@ interface SplitResult<ResponseT, ResultT>
          */
         CompletableFuture<ResultT> resultFuture();
 
+        /**
+         * Indicates if the split async response transformer supports sending individual transformer non-serially, as well as
+         * receiving back data from the many {@link AsyncResponseTransformer#onStream(SdkPublisher) publishers} non-serially.
+         * @return true if non-serial data is supported, false otherwise
+         */
+        default Boolean supportsNonSerial() {
+            return false;
+        }
+
         static <ResponseT, ResultT> Builder<ResponseT, ResultT> builder() {
             return DefaultAsyncResponseTransformerSplitResult.builder();
         }
@@ -380,6 +389,20 @@ interface Builder<ResponseT, ResultT>
              * @return an instance of this Builder
              */
             Builder<ResponseT, ResultT> resultFuture(CompletableFuture<ResultT> future);
+
+            /**
+             * If the AsyncResponseTransformers returned by the {@link SplitResult#publisher()} support concurrent
+             * parallel streaming of multiple content body concurrently.
+             * @return
+             */
+            Boolean supportsNonSerial();
+
+            /**
+             * Sets whether the AsyncResponseTransformers returned by the {@link SplitResult#publisher()} support concurrent
+             * parallel streaming of multiple content body concurrently
+             * @return
+             */
+            Builder<ResponseT, ResultT> supportsNonSerial(Boolean supportsParallel);
         }
     }
 
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/listener/AsyncResponseTransformerListener.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/listener/AsyncResponseTransformerListener.java
index c7ee37690ca0..f1612e1b1f3e 100644
--- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/listener/AsyncResponseTransformerListener.java
+++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/listener/AsyncResponseTransformerListener.java
@@ -20,6 +20,7 @@
 import org.reactivestreams.Subscriber;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.annotations.SdkProtectedApi;
+import software.amazon.awssdk.core.SplittingTransformerConfiguration;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.utils.Logger;
@@ -108,6 +109,11 @@ public String name() {
             return delegate.name();
         }
 
+        @Override
+        public SplitResult<ResponseT, ResultT> split(SplittingTransformerConfiguration splitConfig) {
+            return delegate.split(splitConfig);
+        }
+
         static void invoke(Runnable runnable, String callbackName) {
             try {
                 runnable.run();
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/DefaultAsyncResponseTransformerSplitResult.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/DefaultAsyncResponseTransformerSplitResult.java
index ed64b1d8eae4..8accd9355f85 100644
--- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/DefaultAsyncResponseTransformerSplitResult.java
+++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/DefaultAsyncResponseTransformerSplitResult.java
@@ -27,12 +27,14 @@ public final class DefaultAsyncResponseTransformerSplitResult<ResponseT, ResultT
 
     private final SdkPublisher<AsyncResponseTransformer<ResponseT, ResponseT>> publisher;
     private final CompletableFuture<ResultT> future;
+    private final Boolean supportsNonSerial;
 
     private DefaultAsyncResponseTransformerSplitResult(Builder<ResponseT, ResultT> builder) {
         this.publisher = Validate.paramNotNull(
             builder.publisher(), "asyncResponseTransformerPublisher");
         this.future = Validate.paramNotNull(
             builder.resultFuture(), "future");
+        this.supportsNonSerial = Validate.getOrDefault(builder.supportsNonSerial(), () -> false);
     }
 
     /**
@@ -52,6 +54,11 @@ public CompletableFuture<ResultT> resultFuture() {
         return this.future;
     }
 
+    @Override
+    public Boolean supportsNonSerial() {
+        return this.supportsNonSerial;
+    }
+
     @Override
     public AsyncResponseTransformer.SplitResult.Builder<ResponseT, ResultT> toBuilder() {
         return new DefaultBuilder<>(this);
@@ -65,6 +72,7 @@ public static class DefaultBuilder<ResponseT, ResultT>
         implements AsyncResponseTransformer.SplitResult.Builder<ResponseT, ResultT> {
         private SdkPublisher<AsyncResponseTransformer<ResponseT, ResponseT>> publisher;
         private CompletableFuture<ResultT> future;
+        private Boolean supportsNonSerial;
 
         DefaultBuilder() {
         }
@@ -72,6 +80,7 @@ public static class DefaultBuilder<ResponseT, ResultT>
         DefaultBuilder(DefaultAsyncResponseTransformerSplitResult<ResponseT, ResultT> split) {
             this.publisher = split.publisher;
             this.future = split.future;
+            this.supportsNonSerial = split.supportsNonSerial;
         }
 
         @Override
@@ -92,14 +101,28 @@ public CompletableFuture<ResultT> resultFuture() {
         }
 
         @Override
-        public AsyncResponseTransformer.SplitResult.Builder<ResponseT, ResultT> resultFuture(CompletableFuture<ResultT> future) {
+        public AsyncResponseTransformer.SplitResult.Builder<ResponseT, ResultT> resultFuture(
+            CompletableFuture<ResultT> future) {
             this.future = future;
             return this;
         }
 
+        @Override
+        public Boolean supportsNonSerial() {
+            return supportsNonSerial;
+        }
+
+        @Override
+        public AsyncResponseTransformer.SplitResult.Builder<ResponseT, ResultT> supportsNonSerial(
+            Boolean supportsParallel) {
+            this.supportsNonSerial = supportsParallel;
+            return this;
+        }
+
         @Override
         public AsyncResponseTransformer.SplitResult<ResponseT, ResultT> build() {
             return new DefaultAsyncResponseTransformerSplitResult<>(this);
         }
+
     }
 }
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodySplitHelper.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodySplitHelper.java
index 8cb7d28c6620..92aba8d4762a 100644
--- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodySplitHelper.java
+++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodySplitHelper.java
@@ -28,6 +28,7 @@
 import software.amazon.awssdk.core.async.AsyncRequestBodySplitConfiguration;
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.utils.Logger;
 import software.amazon.awssdk.utils.NumericUtils;
 import software.amazon.awssdk.utils.Validate;
 import software.amazon.awssdk.utils.async.SimplePublisher;
@@ -39,6 +40,7 @@
  */
 @SdkInternalApi
 public final class FileAsyncRequestBodySplitHelper {
+    private static final Logger log = Logger.loggerFor(FileAsyncRequestBodySplitHelper.class);
 
     private final AtomicBoolean isSendingRequestBody = new AtomicBoolean(false);
     private final AtomicLong remainingBytes;
@@ -70,6 +72,9 @@ public FileAsyncRequestBodySplitHelper(FileAsyncRequestBody asyncRequestBody,
                                splitConfiguration.bufferSizeInBytes();
         this.bufferPerAsyncRequestBody = Math.min(asyncRequestBody.chunkSizeInBytes(),
                                                   NumericUtils.saturatedCast(totalBufferSize));
+        log.debug(() -> "chunkSize is " + chunkSize);
+        log.debug(() -> "Total buffer size is " + totalBufferSize);
+        log.debug(() -> "bufferPerAsyncRequestBody is " + bufferPerAsyncRequestBody);
     }
 
     public SdkPublisher<AsyncRequestBody> split() {
@@ -101,9 +106,12 @@ private void sendAsyncRequestBody(SimplePublisher<AsyncRequestBody> simplePublis
 
     private void doSendAsyncRequestBody(SimplePublisher<AsyncRequestBody> simplePublisher) {
         while (shouldSendMore()) {
-            AsyncRequestBody currentAsyncRequestBody = newFileAsyncRequestBody(simplePublisher);
+            FileAsyncRequestBodyWrapper currentAsyncRequestBody = newFileAsyncRequestBody(simplePublisher);
+            long pos = currentAsyncRequestBody.fileAsyncRequestBody.position();
+            log.info(() -> "Sending FileAsyncRequestBodyWrapper for pos:" + pos);
             simplePublisher.send(currentAsyncRequestBody);
             numAsyncRequestBodiesInFlight.incrementAndGet();
+            log.info(() -> "currently in flight after increment: " + numAsyncRequestBodiesInFlight.get());
             checkCompletion(simplePublisher, currentAsyncRequestBody);
         }
     }
@@ -122,11 +130,13 @@ private void checkCompletion(SimplePublisher<AsyncRequestBody> simplePublisher,
     }
 
     private void startNextRequestBody(SimplePublisher<AsyncRequestBody> simplePublisher) {
+        log.info(() -> "Starting next request (decrementing in flight)");
         numAsyncRequestBodiesInFlight.decrementAndGet();
+        log.info(() -> "Currently in flight after decrement: " + numAsyncRequestBodiesInFlight.get());
         sendAsyncRequestBody(simplePublisher);
     }
 
-    private AsyncRequestBody newFileAsyncRequestBody(SimplePublisher<AsyncRequestBody> simplePublisher) {
+    private FileAsyncRequestBodyWrapper newFileAsyncRequestBody(SimplePublisher<AsyncRequestBody> simplePublisher) {
         long position = chunkSize * chunkIndex.getAndIncrement();
         long numBytesToReadForThisChunk = Math.min(totalContentLength - position, chunkSize);
         FileAsyncRequestBody fileAsyncRequestBody = FileAsyncRequestBody.builder()
@@ -145,7 +155,6 @@ private boolean shouldSendMore() {
         if (isDone) {
             return false;
         }
-
         long currentUsedBuffer = (long) numAsyncRequestBodiesInFlight.get() * bufferPerAsyncRequestBody;
         return currentUsedBuffer + bufferPerAsyncRequestBody <= totalBufferSize;
     }
@@ -159,6 +168,7 @@ private final class FileAsyncRequestBodyWrapper implements AsyncRequestBody {
 
         private final FileAsyncRequestBody fileAsyncRequestBody;
         private final SimplePublisher<AsyncRequestBody> simplePublisher;
+        private final AtomicBoolean hasStartedNext = new AtomicBoolean(false);
 
         FileAsyncRequestBodyWrapper(FileAsyncRequestBody fileAsyncRequestBody,
                                     SimplePublisher<AsyncRequestBody> simplePublisher) {
@@ -168,16 +178,28 @@ private final class FileAsyncRequestBodyWrapper implements AsyncRequestBody {
 
         @Override
         public void subscribe(Subscriber<? super ByteBuffer> s) {
-            fileAsyncRequestBody.doAfterOnComplete(() -> startNextRequestBody(simplePublisher))
+            fileAsyncRequestBody.doAfterOnComplete(() -> {
+                                    log.info(() -> "doAfterOnComplete for position " + fileAsyncRequestBody.position());
+                                    startNextRequestBodyOnce();
+                                })
                                 // The reason we still need to call startNextRequestBody when the subscription is
                                 // cancelled is that upstream could cancel the subscription even though the stream has
                                 // finished successfully before onComplete. If this happens, doAfterOnComplete callback
                                 // will never be invoked, and if the current buffer is full, the publisher will stop
                                 // sending new FileAsyncRequestBody, leading to uncompleted future.
-                                .doAfterOnCancel(() -> startNextRequestBody(simplePublisher))
+                                .doAfterOnCancel(() -> {
+                                    log.info(() -> "doAfterOnCancel for position " + fileAsyncRequestBody.position());
+                                    startNextRequestBodyOnce();
+                                })
                                 .subscribe(s);
         }
 
+        private void startNextRequestBodyOnce() {
+            if (hasStartedNext.compareAndSet(false, true)) {
+                startNextRequestBody(simplePublisher);
+            }
+        }
+
         @Override
         public Optional<Long> contentLength() {
             return fileAsyncRequestBody.contentLength();
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformer.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformer.java
index 4348355fa5d8..fc6dd4ad5831 100644
--- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformer.java
+++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformer.java
@@ -41,6 +41,7 @@
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.core.FileTransformerConfiguration;
 import software.amazon.awssdk.core.FileTransformerConfiguration.FailureBehavior;
+import software.amazon.awssdk.core.SplittingTransformerConfiguration;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.core.exception.SdkClientException;
@@ -76,6 +77,18 @@ private FileAsyncResponseTransformer(Path path, FileTransformerConfiguration fil
         this.position = position;
     }
 
+    FileTransformerConfiguration config() {
+        return configuration.toBuilder().build();
+    }
+
+    Path path() {
+        return path;
+    }
+
+    Long initialPosition() {
+        return position;
+    }
+
     private static long determineFilePositionToWrite(Path path, FileTransformerConfiguration fileConfiguration) {
         if (fileConfiguration.fileWriteOption() == CREATE_OR_APPEND_TO_EXISTING) {
             try {
@@ -89,7 +102,7 @@ private static long determineFilePositionToWrite(Path path, FileTransformerConfi
         if (fileConfiguration.fileWriteOption() == WRITE_TO_POSITION) {
             return Validate.getOrDefault(fileConfiguration.position(), () -> 0L);
         }
-        return  0L;
+        return 0L;
     }
 
     private AsynchronousFileChannel createChannel(Path path) throws IOException {
@@ -134,6 +147,10 @@ public void onResponse(ResponseT response) {
         this.response = response;
     }
 
+    public FileTransformerConfiguration getConfiguration() {
+        return this.configuration;
+    }
+
     @Override
     public void onStream(SdkPublisher<ByteBuffer> publisher) {
         try {
@@ -162,6 +179,7 @@ public void exceptionOccurred(Throwable throwable) {
             }
         }
         if (cf != null) {
+            log.info(() -> "completing file future exceptionally");
             cf.completeExceptionally(throwable);
         } else {
             log.warn(() -> "An exception occurred before the call to prepare() was able to instantiate the CompletableFuture."
@@ -174,22 +192,20 @@ public String name() {
         return TransformerType.FILE.getName();
     }
 
-    /**
-     * {@link Subscriber} implementation that writes chunks to a file.
-     */
     static class FileSubscriber implements Subscriber<ByteBuffer> {
         private final AtomicLong position;
         private final AsynchronousFileChannel fileChannel;
         private final Path path;
         private final CompletableFuture<Void> future;
         private final Consumer<Throwable> onErrorMethod;
+        private final Object closeLock = new Object();
 
         private volatile boolean writeInProgress = false;
         private volatile boolean closeOnLastWrite = false;
         private Subscription subscription;
 
         FileSubscriber(AsynchronousFileChannel fileChannel, Path path, CompletableFuture<Void> future,
-                       Consumer<Throwable> onErrorMethod, long startingPosition) {
+                              Consumer<Throwable> onErrorMethod, long startingPosition) {
             this.fileChannel = fileChannel;
             this.path = path;
             this.future = future;
@@ -228,7 +244,7 @@ public void completed(Integer result, ByteBuffer attachment) {
                     if (byteBuffer.hasRemaining()) {
                         performWrite(byteBuffer);
                     } else {
-                        synchronized (FileSubscriber.this) {
+                        synchronized (closeLock) {
                             writeInProgress = false;
                             if (closeOnLastWrite) {
                                 close();
@@ -256,7 +272,7 @@ public void onError(Throwable t) {
         public void onComplete() {
             log.trace(() -> "onComplete");
             // if write in progress, tell write to close on finish.
-            synchronized (this) {
+            synchronized (closeLock) {
                 if (writeInProgress) {
                     log.trace(() -> "writeInProgress = true, not closing");
                     closeOnLastWrite = true;
@@ -284,4 +300,19 @@ public String toString() {
             return getClass() + ":" + path.toString();
         }
     }
+
+
+    @Override
+    public SplitResult<ResponseT, ResponseT> split(SplittingTransformerConfiguration splitConfig) {
+        if (configuration.fileWriteOption() == CREATE_OR_APPEND_TO_EXISTING) {
+            return AsyncResponseTransformer.super.
+                split(splitConfig)
+                .copy(res -> res.supportsNonSerial(false));
+        }
+        return AsyncResponseTransformer.super
+            .split(splitConfig)
+            .copy(
+                res -> res.supportsNonSerial(true)
+                          .publisher(new FileAsyncResponseTransformerPublisher(this)));
+    }
 }
\ No newline at end of file
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java
new file mode 100644
index 000000000000..45d04b7b9963
--- /dev/null
+++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java
@@ -0,0 +1,196 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core.internal.async;
+
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.reactivestreams.Subscriber;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.FileTransformerConfiguration;
+import software.amazon.awssdk.core.SdkResponse;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.core.async.SdkPublisher;
+import software.amazon.awssdk.utils.CollectionUtils;
+import software.amazon.awssdk.utils.CompletableFutureUtils;
+import software.amazon.awssdk.utils.ContentRangeParser;
+import software.amazon.awssdk.utils.Logger;
+import software.amazon.awssdk.utils.Pair;
+import software.amazon.awssdk.utils.StringUtils;
+import software.amazon.awssdk.utils.Validate;
+
+/**
+ * A publisher of {@link FileAsyncResponseTransformer} that uses the Content-Range header of a {@link SdkResponse} to write to the
+ * offset defined in the range of the Content-Range. Correspond to the {@link SplittingTransformer} for non-linear write cases.
+ */
+@SdkInternalApi
+public class FileAsyncResponseTransformerPublisher<T extends SdkResponse>
+    implements SdkPublisher<AsyncResponseTransformer<T, T>> {
+    private static final Logger log = Logger.loggerFor(FileAsyncResponseTransformerPublisher.class);
+
+    private final Path path;
+    private final FileTransformerConfiguration initialConfig;
+    private final long initialPosition;
+    private final AtomicLong amountOfTransformerSent;
+    private Subscriber<?> subscriber;
+
+    public FileAsyncResponseTransformerPublisher(FileAsyncResponseTransformer<?> responseTransformer) {
+        this.path = Validate.paramNotNull(responseTransformer.path(), "path");
+        this.initialConfig = Validate.paramNotNull(responseTransformer.config(), "fileTransformerConfiguration");
+        this.initialPosition = responseTransformer.initialPosition();
+        this.amountOfTransformerSent = new AtomicLong(0);
+    }
+
+    @Override
+    public void subscribe(Subscriber<? super AsyncResponseTransformer<T, T>> s) {
+        if (s == null) {
+            throw new NullPointerException("Subscription must not be null");
+        }
+        this.subscriber = s;
+        s.onSubscribe(new ThreadSafeEmittingSubscription<>(
+            s,
+            new AtomicLong(0),
+            this::onCancel,
+            new AtomicBoolean(),
+            log,
+            this::createTransformer));
+    }
+
+    private AsyncResponseTransformer<T, T> createTransformer() {
+        return new IndividualFileTransformer();
+    }
+
+    private void onCancel() {
+        subscriber = null;
+    }
+
+    /**
+     * This is the AsyncResponseTransformer that will be used for each individual requests.
+     * <p>
+     * We delegate to new instances of the already existing class {@link FileAsyncResponseTransformer} to perform the individual
+     * requests. This FileAsyncResponseTransformer will write the content of the request to the file at the offset taken from the
+     * Content-Range header ('x-amz-content-range'). As such, we don't need to manually manage the state of the
+     * AsyncResponseTransformer passed by the user, like we do for {@link SplittingTransformer}. Here, we know it is a
+     * FileAsyncResponseTransformer, so we can just ignore it, and instead rely on the individual FileAsyncResponseTransformer of
+     * every part.
+     * <p>
+     * Note on retries: since we are delegating requests to {@link FileAsyncResponseTransformer}, each request made with this
+     * transformer will retry independently based on the retry configuration of the client it is used with. We only need to verify
+     * the completion state of the future of each individually
+     */
+    private class IndividualFileTransformer implements AsyncResponseTransformer<T, T> {
+        private AsyncResponseTransformer<T, T> delegate;
+        private CompletableFuture<T> future;
+
+        @Override
+        public CompletableFuture<T> prepare() {
+            this.future = new CompletableFuture<>();
+            return this.future;
+        }
+
+        @Override
+        public void onResponse(T response) {
+            List<String> contentRangeList;
+            contentRangeList = response.sdkHttpResponse().headers().get("x-amz-content-range");
+            if (CollectionUtils.isNullOrEmpty(contentRangeList) || StringUtils.isEmpty(contentRangeList.get(0))) {
+                contentRangeList = response.sdkHttpResponse().headers().get("content-range");
+                if (CollectionUtils.isNullOrEmpty(contentRangeList) || StringUtils.isEmpty(contentRangeList.get(0))) {
+                    // Bad state! This is intended to cancel everything
+                    if (subscriber != null) {
+                        subscriber.onError(new IllegalStateException("Content range header is missing"));
+                    }
+                    return;
+                }
+            }
+
+            String contentRange = contentRangeList.get(0);
+            Optional<Pair<Long, Long>> contentRangePair = ContentRangeParser.range(contentRange);
+            if (!contentRangePair.isPresent()) {
+                // Bad state! This is intended to cancel everything
+                if (subscriber != null) {
+                    subscriber.onError(new IllegalStateException("Could not parse content range header " + contentRange));
+                }
+                return;
+            }
+
+            this.delegate = getDelegateTransformer(contentRangePair.get().left());
+            CompletableFuture<T> delegateFuture = delegate.prepare();
+            CompletableFutureUtils.forwardResultTo(delegateFuture, future);
+            CompletableFutureUtils.forwardExceptionTo(future, delegateFuture);
+            amountOfTransformerSent.incrementAndGet();
+            delegate.onResponse(response);
+        }
+
+        private AsyncResponseTransformer<T, T> getDelegateTransformer(Long startAt) {
+            if (amountOfTransformerSent.get() == 0) {
+                return AsyncResponseTransformer.toFile(path, initialConfig);
+            }
+            switch (initialConfig.fileWriteOption()) {
+                case CREATE_NEW:
+                case CREATE_OR_REPLACE_EXISTING: {
+                    FileTransformerConfiguration newConfig = initialConfig.copy(c -> c
+                        .fileWriteOption(FileTransformerConfiguration.FileWriteOption.WRITE_TO_POSITION)
+                        .position(startAt));
+                    return AsyncResponseTransformer.toFile(path, newConfig);
+                }
+                case CREATE_OR_APPEND_TO_EXISTING: {
+                    // APPEND mode is not supported for non-serial operations,
+                    // we can just reuse the same config (with CREATE_OR_APPEND_TO_EXISTING) which will work serially
+                    return AsyncResponseTransformer.toFile(path, initialConfig);
+                }
+                case WRITE_TO_POSITION: {
+                    long initialOffset = initialConfig.position();
+                    FileTransformerConfiguration newConfig = initialConfig.copy(c -> c
+                        .fileWriteOption(FileTransformerConfiguration.FileWriteOption.WRITE_TO_POSITION)
+                        .position(initialOffset + startAt));
+                    return AsyncResponseTransformer.toFile(path, newConfig);
+                }
+                default:
+                    throw new UnsupportedOperationException("Unsupported fileWriteOption: " + initialConfig.fileWriteOption());
+            }
+        }
+
+        @Override
+        public void onStream(SdkPublisher<ByteBuffer> publisher) {
+            if (delegate != null) {
+                // should never be null as per AsyncResponseTransformer runtime contract, but we never know
+                delegate.onStream(publisher);
+            }
+        }
+
+        @Override
+        public void exceptionOccurred(Throwable error) {
+            log.error(() -> "error detected", error);
+            if (delegate != null) {
+                // do not call onError, because exceptionOccurred may be called multiple times due to retries, simply forward the
+                // error to the delegate async response transformer which will let the service call pipeline handle the error.
+                delegate.exceptionOccurred(error);
+            } else {
+                // If we received an error without even having a delegate, this means we have thrown an error before even
+                // getting a onResponse signal. We complete the prepared future, to let the
+                // service call pipeline handle the error
+                if (future != null) {
+                    future.completeExceptionally(error);
+                }
+            }
+        }
+    }
+
+}
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileSplittingTransformer.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileSplittingTransformer.java
new file mode 100644
index 000000000000..eb4a5b21b193
--- /dev/null
+++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileSplittingTransformer.java
@@ -0,0 +1,19 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core.internal.async;
+
+public class FileSplittingTransformer {
+}
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ThreadSafeEmittingSubscription.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ThreadSafeEmittingSubscription.java
new file mode 100644
index 000000000000..6cff201e7a8a
--- /dev/null
+++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ThreadSafeEmittingSubscription.java
@@ -0,0 +1,101 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core.internal.async;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.utils.Logger;
+
+@SdkInternalApi
+public class ThreadSafeEmittingSubscription<T> implements Subscription {
+
+    private final AtomicBoolean emitting = new AtomicBoolean(false);
+    private final AtomicLong outstandingDemand;
+    private final Runnable onCancel;
+    private final AtomicBoolean isCancelled;
+    private final Supplier<T> supplier;
+    private final Logger log;
+
+    private Subscriber<T> downstreamSubscriber;
+
+    public ThreadSafeEmittingSubscription(Subscriber<T> downstreamSubscriber, AtomicLong outstandingDemand,
+                                          Runnable onCancel, AtomicBoolean isCancelled, Logger log, Supplier<T> supplier) {
+        this.downstreamSubscriber = downstreamSubscriber;
+        this.outstandingDemand = outstandingDemand;
+        this.onCancel = onCancel;
+        this.isCancelled = isCancelled;
+        this.log = log;
+        this.supplier = supplier;
+    }
+
+    @Override
+    public void request(long n) {
+        if (n <= 0) {
+            downstreamSubscriber.onError(new IllegalArgumentException("Amount requested must be positive"));
+            return;
+        }
+        long newDemand = outstandingDemand.updateAndGet(current -> {
+            if (Long.MAX_VALUE - current < n) {
+                return Long.MAX_VALUE;
+            }
+            return current + n;
+        });
+        log.trace(() -> String.format("new outstanding demand: %s", newDemand));
+        emit();
+    }
+
+    @Override
+    public void cancel() {
+        isCancelled.set(true);
+        downstreamSubscriber = null;
+        onCancel.run();
+    }
+
+    private void emit() {
+        do {
+            if (!emitting.compareAndSet(false, true)) {
+                return;
+            }
+            try {
+                if (doEmit()) {
+                    return;
+                }
+            } finally {
+                emitting.compareAndSet(true, false);
+            }
+        } while (outstandingDemand.get() > 0);
+    }
+
+    private boolean doEmit() {
+        long demand = outstandingDemand.get();
+
+        while (demand > 0) {
+            if (isCancelled.get()) {
+                return true;
+            }
+            if (outstandingDemand.get() > 0) {
+                demand = outstandingDemand.decrementAndGet();
+                downstreamSubscriber.onNext(supplier.get());
+            }
+        }
+        return false;
+    }
+
+}
diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisherTckTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisherTckTest.java
new file mode 100644
index 000000000000..6f8c1616dda5
--- /dev/null
+++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisherTckTest.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core.internal.async;
+
+import com.google.common.jimfs.Configuration;
+import com.google.common.jimfs.Jimfs;
+import java.nio.file.FileSystem;
+import java.nio.file.Path;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.tck.PublisherVerification;
+import org.reactivestreams.tck.TestEnvironment;
+import software.amazon.awssdk.core.SdkResponse;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.core.async.SdkPublisher;
+
+public class FileAsyncResponseTransformerPublisherTckTest extends PublisherVerification<AsyncResponseTransformer<SdkResponse,
+    SdkResponse>> {
+
+    private final FileSystem fileSystem = Jimfs.newFileSystem(Configuration.unix());
+    private final Path testFile = fileSystem.getPath("/test-file.txt");
+
+
+    public FileAsyncResponseTransformerPublisherTckTest() {
+        super(new TestEnvironment());
+    }
+
+    @Override
+    public Publisher<AsyncResponseTransformer<SdkResponse, SdkResponse>> createPublisher(long elements) {
+        FileAsyncResponseTransformer<?> art =
+            (FileAsyncResponseTransformer<?>) AsyncResponseTransformer.toFile(testFile);
+        FileAsyncResponseTransformerPublisher<SdkResponse> publisher =
+            new FileAsyncResponseTransformerPublisher<>(art);
+
+        return SdkPublisher.adapt(publisher).limit((int) elements);
+    }
+
+    @Override
+    public Publisher<AsyncResponseTransformer<SdkResponse, SdkResponse>> createFailedPublisher() {
+        return null;
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return Long.MAX_VALUE;
+    }
+
+}
\ No newline at end of file
diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisherTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisherTest.java
new file mode 100644
index 000000000000..42fa4d939ce6
--- /dev/null
+++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisherTest.java
@@ -0,0 +1,249 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core.internal.async;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.jimfs.Configuration;
+import com.google.common.jimfs.Jimfs;
+import java.nio.ByteBuffer;
+import java.nio.file.FileSystem;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import software.amazon.awssdk.core.SdkResponse;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.core.async.SdkPublisher;
+import software.amazon.awssdk.http.SdkHttpResponse;
+import software.amazon.awssdk.utils.CompletableFutureUtils;
+
+class FileAsyncResponseTransformerPublisherTest {
+
+    private FileSystem fileSystem;
+    private Path testFile;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        fileSystem = Jimfs.newFileSystem(Configuration.unix());
+        testFile = fileSystem.getPath("/test-file.txt");
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        fileSystem.close();
+    }
+
+    @Test
+    void happyPath_singleOnNext() throws Exception {
+        // Given
+        AsyncResponseTransformer<Object, Object> initialTransformer = AsyncResponseTransformer.toFile(testFile);
+
+        FileAsyncResponseTransformerPublisher<SdkResponse> publisher =
+            new FileAsyncResponseTransformerPublisher<>((FileAsyncResponseTransformer<?>) initialTransformer);
+
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<AsyncResponseTransformer<SdkResponse, SdkResponse>> receivedTransformer = new AtomicReference<>();
+        CompletableFuture<SdkResponse> future = new CompletableFuture<>();
+
+        // When
+        publisher.subscribe(new Subscriber<AsyncResponseTransformer<SdkResponse, SdkResponse>>() {
+            private Subscription subscription;
+
+            @Override
+            public void onSubscribe(Subscription s) {
+                this.subscription = s;
+                s.request(1);
+            }
+
+            @Override
+            public void onNext(AsyncResponseTransformer<SdkResponse, SdkResponse> transformer) {
+                receivedTransformer.set(transformer);
+
+                // Simulate response with content-range header
+                SdkResponse mockResponse = createMockResponse();
+                CompletableFuture<SdkResponse> prepareFuture = transformer.prepare();
+                CompletableFutureUtils.forwardResultTo(prepareFuture, future);
+                transformer.onResponse(mockResponse);
+
+                // Simulate stream data
+                SdkPublisher<ByteBuffer> mockPublisher = createMockPublisher();
+                transformer.onStream(mockPublisher);
+
+                latch.countDown();
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                latch.countDown();
+            }
+
+            @Override
+            public void onComplete() {
+                latch.countDown();
+            }
+        });
+
+        // Then
+        assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+        assertThat(receivedTransformer.get()).isNotNull();
+        assertThat(Files.exists(testFile)).isTrue();
+        assertThat(future).isCompleted();
+    }
+
+    private SdkResponse createMockResponse() {
+        SdkResponse mockResponse = mock(SdkResponse.class);
+        SdkHttpResponse mockHttpResponse = mock(SdkHttpResponse.class);
+
+        when(mockResponse.sdkHttpResponse()).thenReturn(mockHttpResponse);
+        when(mockHttpResponse.headers()).thenReturn(
+            Collections.singletonMap("x-amz-content-range", Collections.singletonList("bytes 0-9/10"))
+        );
+
+        return mockResponse;
+    }
+
+    private SdkPublisher<ByteBuffer> createMockPublisher() {
+        return s -> s.onSubscribe(new Subscription() {
+            @Override
+            public void request(long n) {
+                s.onNext(ByteBuffer.wrap("test data".getBytes()));
+                s.onComplete();
+            }
+
+            @Override
+            public void cancel() {
+            }
+        });
+    }
+
+    @Test
+    void multipleOnNext_differentContentRanges() throws Exception {
+        // Given
+        AsyncResponseTransformer<Object, Object> initialTransformer = AsyncResponseTransformer.toFile(testFile);
+        FileAsyncResponseTransformerPublisher<SdkResponse> publisher =
+            new FileAsyncResponseTransformerPublisher<>((FileAsyncResponseTransformer<?>) initialTransformer);
+
+        int numTransformers = 8;
+        CountDownLatch latch = new CountDownLatch(numTransformers);
+        AtomicInteger transformerCount = new AtomicInteger(0);
+        List<CompletableFuture<SdkResponse>> futures = new ArrayList<>();
+
+        // When
+        publisher.subscribe(new Subscriber<AsyncResponseTransformer<SdkResponse, SdkResponse>>() {
+            @Override
+            public void onSubscribe(Subscription s) {
+                s.request(numTransformers);
+            }
+
+            @Override
+            public void onNext(AsyncResponseTransformer<SdkResponse, SdkResponse> transformer) {
+                int index = transformerCount.getAndIncrement();
+
+                // Each transformer gets a different 10-byte range
+                long startByte = index * 10L;
+                long endByte = startByte + 9;
+                String contentRange = String.format("bytes %d-%d/80", startByte, endByte);
+                byte[] data = new byte[10];
+                for (int i = 0; i < 10; i++) {
+                    data[i] = (byte) ((byte) startByte + i);
+                }
+
+                SdkResponse mockResponse = createMockResponseWithRange(contentRange);
+                CompletableFuture<SdkResponse> future = transformer.prepare();
+                futures.add(future);
+
+                transformer.onResponse(mockResponse);
+                transformer.onStream(createMockPublisherWithData(data));
+
+                latch.countDown();
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                for (int i = 0; i < numTransformers; i++) {
+                    latch.countDown();
+                }
+            }
+
+            @Override
+            public void onComplete() {
+            }
+        });
+
+        // Then
+        assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+        assertThat(transformerCount.get()).isEqualTo(numTransformers);
+
+        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+
+        assertThat(Files.exists(testFile)).isTrue();
+        byte[] fileContent = Files.readAllBytes(testFile);
+
+        assertThat(fileContent.length).isEqualTo(80);
+
+        for (int i = 0; i < numTransformers; i++) {
+            int startPos = i * 10;
+            byte[] expectedData = new byte[10];
+            for (int j = 0; j < 10; j++) {
+                expectedData[j] = (byte) ((byte) startPos + j);
+            }
+            byte[] actualData = Arrays.copyOfRange(fileContent, startPos, startPos + 10);
+            assertThat(actualData).isEqualTo(expectedData);
+        }
+    }
+
+    private SdkResponse createMockResponseWithRange(String contentRange) {
+        SdkResponse mockResponse = mock(SdkResponse.class);
+        SdkHttpResponse mockHttpResponse = mock(SdkHttpResponse.class);
+
+        when(mockResponse.sdkHttpResponse()).thenReturn(mockHttpResponse);
+        when(mockHttpResponse.headers()).thenReturn(
+            Collections.singletonMap("x-amz-content-range", Collections.singletonList(contentRange))
+        );
+
+        return mockResponse;
+    }
+
+    private SdkPublisher<ByteBuffer> createMockPublisherWithData(byte[] data) {
+        return s -> s.onSubscribe(new Subscription() {
+            @Override
+            public void request(long n) {
+                s.onNext(ByteBuffer.wrap(data));
+                s.onComplete();
+            }
+
+            @Override
+            public void cancel() {
+            }
+        });
+    }
+
+}
\ No newline at end of file
diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileSubscriberTckTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileSubscriberTckTest.java
index 8810c3459f25..1937de5220e1 100644
--- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileSubscriberTckTest.java
+++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileSubscriberTckTest.java
@@ -31,10 +31,9 @@
 import org.reactivestreams.Subscription;
 import org.reactivestreams.tck.SubscriberWhiteboxVerification;
 import org.reactivestreams.tck.TestEnvironment;
-import software.amazon.awssdk.core.internal.async.FileAsyncResponseTransformer.FileSubscriber;
 
 /**
- * TCK verification test for {@link FileSubscriber}.
+ * TCK verification test for {@link FileAsyncResponseTransformer.FileSubscriber}.
  */
 public class FileSubscriberTckTest extends SubscriberWhiteboxVerification<ByteBuffer> {
     private static final byte[] CONTENT = new byte[16];
@@ -47,7 +46,7 @@ public FileSubscriberTckTest() {
     @Override
     public Subscriber<ByteBuffer> createSubscriber(WhiteboxSubscriberProbe<ByteBuffer> whiteboxSubscriberProbe) {
         Path tempFile = getNewTempFile();
-        return new FileSubscriber(openChannel(tempFile), tempFile, new CompletableFuture<>(), (t) -> {}, 0) {
+        return new FileAsyncResponseTransformer.FileSubscriber(openChannel(tempFile), tempFile, new CompletableFuture<>(), (t) -> {}, 0) {
             @Override
             public void onSubscribe(Subscription s) {
                 super.onSubscribe(s);
diff --git a/core/sdk-core/src/test/resources/log4j2.properties b/core/sdk-core/src/test/resources/log4j2.properties
index e5e68dd2faa4..d356ebf326c1 100644
--- a/core/sdk-core/src/test/resources/log4j2.properties
+++ b/core/sdk-core/src/test/resources/log4j2.properties
@@ -23,6 +23,9 @@ appender.console.layout.pattern = %d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %
 rootLogger.level = info
 rootLogger.appenderRef.stdout.ref = ConsoleAppender
 
+logger.upload.name = software.amazon.awssdk.services.s3.internal.multipart.UploadWithKnownContentLengthHelper
+logger.upload.level = debug
+
 # Uncomment below to enable more specific logging
 #
 #logger.sdk.name = software.amazon.awssdk
diff --git a/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/GenericS3TransferManager.java b/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/GenericS3TransferManager.java
index 29bcdc34d93f..5a6ea0569bbb 100644
--- a/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/GenericS3TransferManager.java
+++ b/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/GenericS3TransferManager.java
@@ -341,10 +341,19 @@ public final <ResultT> Download<ResultT> download(DownloadRequest<ResultT> downl
 
         TransferProgressUpdater progressUpdater = new TransferProgressUpdater(downloadRequest, null);
         progressUpdater.transferInitiated();
-        responseTransformer = isS3ClientMultipartEnabled()
-                              ? progressUpdater.wrapResponseTransformerForMultipartDownload(
-            responseTransformer, downloadRequest.getObjectRequest())
-                              : progressUpdater.wrapResponseTransformer(responseTransformer);
+        if (isS3ClientMultipartEnabled()) {
+            // TODO find a better way to do that
+            if (responseTransformer.split(b -> b.bufferSizeInBytes(1L)).supportsNonSerial()) {
+                responseTransformer =
+                    progressUpdater.wrapForNonSerialFileDownload(responseTransformer, downloadRequest.getObjectRequest());
+            } else {
+                responseTransformer =
+                    progressUpdater.wrapResponseTransformerForMultipartDownload(responseTransformer,
+                                                                                downloadRequest.getObjectRequest());
+            }
+        } else {
+            responseTransformer = progressUpdater.wrapResponseTransformer(responseTransformer);
+        }
         progressUpdater.registerCompletion(returnFuture);
 
         try {
@@ -374,7 +383,8 @@ public final FileDownload downloadFile(DownloadFileRequest downloadRequest) {
             downloadRequest.getObjectRequest(),
             b -> b.putExecutionAttribute(MULTIPART_DOWNLOAD_RESUME_CONTEXT, new MultipartDownloadResumeContext()));
         DownloadFileRequest downloadFileRequestWithAttributes =
-            downloadRequest.copy(downloadFileRequest -> downloadFileRequest.getObjectRequest(getObjectRequestWithAttributes));
+            downloadRequest.copy(downloadFileRequest ->
+                                     downloadFileRequest.getObjectRequest(getObjectRequestWithAttributes));
 
         AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> responseTransformer =
             AsyncResponseTransformer.toFile(downloadFileRequestWithAttributes.destination(),
@@ -384,7 +394,8 @@ public final FileDownload downloadFile(DownloadFileRequest downloadRequest) {
         TransferProgressUpdater progressUpdater = doDownloadFile(
             downloadFileRequestWithAttributes, responseTransformer, returnFuture);
 
-        return new DefaultFileDownload(returnFuture, progressUpdater.progress(), () -> downloadFileRequestWithAttributes, null);
+        return new DefaultFileDownload(returnFuture, progressUpdater.progress(), () -> downloadFileRequestWithAttributes,
+                                       null);
     }
 
     private TransferProgressUpdater doDownloadFile(
@@ -395,7 +406,7 @@ private TransferProgressUpdater doDownloadFile(
         try {
             progressUpdater.transferInitiated();
             responseTransformer = isS3ClientMultipartEnabled()
-                                  ? progressUpdater.wrapResponseTransformerForMultipartDownload(
+                                  ? progressUpdater.wrapForNonSerialFileDownload(
                 responseTransformer, downloadRequest.getObjectRequest())
                                   : progressUpdater.wrapResponseTransformer(responseTransformer);
             progressUpdater.registerCompletion(returnFuture);
diff --git a/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/progress/TransferProgressUpdater.java b/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/progress/TransferProgressUpdater.java
index 2cab45039d97..4d78530a45ce 100644
--- a/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/progress/TransferProgressUpdater.java
+++ b/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/progress/TransferProgressUpdater.java
@@ -22,8 +22,10 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.reactivestreams.Subscriber;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.SplittingTransformerConfiguration;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.core.async.listener.AsyncRequestBodyListener;
 import software.amazon.awssdk.core.async.listener.AsyncResponseTransformerListener;
 import software.amazon.awssdk.core.async.listener.PublisherListener;
@@ -35,6 +37,7 @@
 import software.amazon.awssdk.transfer.s3.progress.TransferListener;
 import software.amazon.awssdk.transfer.s3.progress.TransferProgress;
 import software.amazon.awssdk.transfer.s3.progress.TransferProgressSnapshot;
+import software.amazon.awssdk.utils.ContentRangeParser;
 
 /**
  * An SDK-internal helper class that facilitates updating a {@link TransferProgress} and invoking {@link TransferListener}s.
@@ -172,25 +175,85 @@ public <ResultT> AsyncResponseTransformer<GetObjectResponse, ResultT> wrapRespon
             new BaseAsyncResponseTransformerListener() {
                 @Override
                 public void transformerOnResponse(GetObjectResponse response) {
-                    // if the GetObjectRequest is a range-get, the Content-Length headers of the response needs to be used
-                    // to update progress since the Content-Range would incorrectly upgrade progress with the whole object
-                    // size.
-                    if (request.range() != null) {
-                        if (response.contentLength() != null) {
-                            progress.updateAndGet(b -> b.totalBytes(response.contentLength()).sdkResponse(response));
-                        }
-                    } else {
-                        // if the GetObjectRequest is not a range-get, it might be a part-get. In that case, we need to parse
-                        // the Content-Range header to get the correct totalByte amount.
-                        ContentRangeParser
-                            .totalBytes(response.contentRange())
-                            .ifPresent(totalBytes -> progress.updateAndGet(b -> b.totalBytes(totalBytes).sdkResponse(response)));
-                    }
+                    multipartDownloadOnResponse(request, response);
                 }
             }
         );
     }
 
+    private void multipartDownloadOnResponse(GetObjectRequest request, GetObjectResponse response) {
+        // if the GetObjectRequest is a range-get, the Content-Length headers of the response needs to be used
+        // to update progress since the Content-Range would incorrectly upgrade progress with the whole object
+        // size.
+        if (request.range() != null) {
+            if (response.contentLength() != null) {
+                progress.updateAndGet(b -> b.totalBytes(response.contentLength()).sdkResponse(response));
+            }
+        } else {
+            // if the GetObjectRequest is not a range-get, it might be a part-get. In that case, we need to parse
+            // the Content-Range header to get the correct totalByte amount.
+            ContentRangeParser
+                .totalBytes(response.contentRange())
+                .ifPresent(totalBytes -> progress.updateAndGet(b -> b.totalBytes(totalBytes).sdkResponse(response)));
+        }
+    }
+
+    // upstream transformer
+    public <ResultT> AsyncResponseTransformer<GetObjectResponse, ResultT> wrapForNonSerialFileDownload(
+        AsyncResponseTransformer<GetObjectResponse, ResultT> responseTransformer, GetObjectRequest request) {
+        return new AsyncResponseTransformer<GetObjectResponse, ResultT>() {
+            @Override
+            public CompletableFuture<ResultT> prepare() {
+                return responseTransformer.prepare();
+            }
+
+            @Override
+            public void onResponse(GetObjectResponse response) {
+                responseTransformer.onResponse(response);
+            }
+
+            @Override
+            public void onStream(SdkPublisher<ByteBuffer> publisher) {
+                responseTransformer.onStream(publisher);
+            }
+
+            @Override
+            public void exceptionOccurred(Throwable error) {
+                responseTransformer.exceptionOccurred(error);
+            }
+
+            @Override
+            public SplitResult<GetObjectResponse, ResultT> split(SplittingTransformerConfiguration splitConfig) {
+                return responseTransformer
+                    .split(splitConfig)
+                    .copy(b -> b.publisher(wrapIndividualTransformer(b.publisher(), request)));
+            }
+
+            @Override
+            public String name() {
+                return responseTransformer.name();
+            }
+        };
+    }
+
+    private <ResultT> SdkPublisher<AsyncResponseTransformer<GetObjectResponse, ResultT>> wrapIndividualTransformer(
+        SdkPublisher<AsyncResponseTransformer<GetObjectResponse, ResultT>> publisher, GetObjectRequest request) {
+        // each of the individual transformer for multipart file download
+        return publisher.map(art -> AsyncResponseTransformerListener.wrap(
+                art,
+                new AsyncResponseTransformerListener<GetObjectResponse>() {
+                    @Override
+                    public void transformerOnResponse(GetObjectResponse response) {
+                        multipartDownloadOnResponse(request, response);
+                    }
+
+                    @Override
+                    public void subscriberOnNext(ByteBuffer byteBuffer) {
+                        incrementBytesTransferred(byteBuffer.limit());
+                    }
+                }));
+    }
+
     public <ResultT> AsyncResponseTransformer<GetObjectResponse, ResultT> wrapResponseTransformer(
         AsyncResponseTransformer<GetObjectResponse, ResultT> responseTransformer) {
         return AsyncResponseTransformerListener.wrap(
diff --git a/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/HagridTest.java b/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/HagridTest.java
new file mode 100644
index 000000000000..53e28c293a70
--- /dev/null
+++ b/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/HagridTest.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.transfer.s3;
+
+import static software.amazon.awssdk.transfer.s3.SizeConstant.GB;
+import static software.amazon.awssdk.transfer.s3.SizeConstant.KB;
+import static software.amazon.awssdk.transfer.s3.SizeConstant.MB;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.transfer.s3.model.CompletedDownload;
+import software.amazon.awssdk.transfer.s3.model.CompletedFileDownload;
+import software.amazon.awssdk.transfer.s3.model.CompletedFileUpload;
+import software.amazon.awssdk.transfer.s3.model.DownloadFileRequest;
+import software.amazon.awssdk.transfer.s3.model.DownloadRequest;
+import software.amazon.awssdk.transfer.s3.model.FileUpload;
+import software.amazon.awssdk.transfer.s3.model.UploadFileRequest;
+import software.amazon.awssdk.transfer.s3.progress.LoggingTransferListener;
+import software.amazon.awssdk.utils.Logger;
+
+public class HagridTest {
+    private static final Logger log = Logger.loggerFor(HagridTest.class);
+
+    int maxInflightDownloads = 8;
+    String bucket = "do-not-delete-java-hagrid-test";
+    long partSize = 5L * 1024 * 1024 * 1024;
+    int chunkSize = 16 * 1024; //16KB
+    long bufferSize = chunkSize * maxInflightDownloads;
+
+    @Test
+    void getHagridFile() throws IOException {
+        String testPath = System.getProperty("testpath");
+        String key = System.getProperty("testkey");
+        S3AsyncClient s3AsyncClient =
+            S3AsyncClient.builder()
+                         .region(Region.US_WEST_2)
+                         .multipartEnabled(true)
+                         .multipartConfiguration(c -> c.maxInflightDownloads(maxInflightDownloads))
+                         .httpClient(NettyNioAsyncHttpClient.builder()
+                                                            .connectionTimeout(Duration.ofMinutes(30))
+                                                            .connectionAcquisitionTimeout(Duration.ofMillis(Integer.MAX_VALUE))
+                                                            .connectionMaxIdleTime(Duration.ofMinutes(2))
+                                                            .build())
+                         .build();
+        Path path = Paths.get(String.format(testPath, System.currentTimeMillis()));
+        S3TransferManager manager = S3TransferManager.builder()
+                                                     .s3Client(s3AsyncClient)
+                                                     .build();
+        CompletableFuture<CompletedFileDownload> fut =
+            manager.downloadFile(DownloadFileRequest.builder()
+                                                    .getObjectRequest(get -> get.key(key)
+                                                                                .bucket(bucket))
+                                                    .destination(path)
+                                                    // .addTransferListener(LoggingTransferListener.create())
+                                                    .build())
+                   .completionFuture();
+        long start = System.currentTimeMillis();
+        CompletedFileDownload res = fut.join();
+        long end = System.currentTimeMillis();
+        System.out.println(res.response());
+        long latencyInSec = (end - start) / 1000;
+        printOutResult(latencyInSec, Files.size(path));
+    }
+
+    @Test
+    void uploadHagridFile() throws IOException {
+        String testPath = System.getProperty("testpath");
+        String key = System.getProperty("testkey");
+        S3AsyncClient s3AsyncClient =
+            S3AsyncClient.builder()
+                         .region(Region.US_WEST_2)
+                         .multipartEnabled(true)
+                         .multipartConfiguration(c -> c
+                             .minimumPartSizeInBytes(partSize)
+                             .apiCallBufferSizeInBytes(bufferSize))
+                         .httpClient(NettyNioAsyncHttpClient.builder()
+                                                            .connectionTimeout(Duration.ofMinutes(30))
+                                                            .connectionAcquisitionTimeout(Duration.ofMillis(Integer.MAX_VALUE))
+                                                            .connectionMaxIdleTime(Duration.ofMinutes(2))
+                                                            .build())
+                         .build();
+        Path path = Paths.get(String.format(testPath, System.currentTimeMillis()));
+        S3TransferManager manager = S3TransferManager.builder()
+                                                     .s3Client(s3AsyncClient)
+                                                     .build();
+
+        FileUpload fileUpload = manager.uploadFile(
+            UploadFileRequest.builder()
+                             .addTransferListener(LoggingTransferListener.create())
+                             .putObjectRequest(
+                                 put -> put.key(key).bucket(bucket))
+                             .source(path)
+                             .build());
+
+        long start = System.currentTimeMillis();
+        CompletedFileUpload upload = fileUpload.completionFuture().join();
+        long end = System.currentTimeMillis();
+        System.out.println(upload.response());
+        long latencyInSec = (end - start) / 1000;
+        printOutResult(latencyInSec, Files.size(path));
+    }
+
+    public static void printOutResult(long latency, long contentLengthInByte) {
+        double contentLengthInGigabit = (contentLengthInByte / (double) GB) * 8.0;
+        System.out.printf("Content Length (Bytes): %d%n", contentLengthInByte);
+        System.out.printf("Average latency (s): %d%n", latency);
+        System.out.printf("Object size (Gigabit): %.4f%n", contentLengthInGigabit);
+        System.out.printf("Average throughput (Gbps): %.4f%n", contentLengthInGigabit / latency);
+    }
+
+}
diff --git a/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/progress/ContentRangeParserTest.java b/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/progress/ContentRangeParserTest.java
index 6dc7a7fc3ce8..b432d93996a7 100644
--- a/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/progress/ContentRangeParserTest.java
+++ b/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/progress/ContentRangeParserTest.java
@@ -17,12 +17,14 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.stream.Stream;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.utils.ContentRangeParser;
+import software.amazon.awssdk.utils.Pair;
 
 class ContentRangeParserTest {
 
@@ -51,4 +53,18 @@ static Stream<Arguments> argumentProvider() {
             Arguments.of("bla bla bla", OptionalLong.empty()));
     }
 
+    @ParameterizedTest
+    @MethodSource("testRange")
+    void testRange(String contentRange, Optional<Pair<Long, Long>> expected) {
+        assertThat(ContentRangeParser.range(contentRange)).isEqualTo(expected);
+    }
+
+    static Stream<Arguments> testRange() {
+        return Stream.of(
+            Arguments.of("bytes 0-9/10", Optional.of(Pair.of(0L, 9L))),
+            Arguments.of("bytes */10", Optional.empty()),
+            Arguments.of("bytes 12000000-17999999/30000000", Optional.of(Pair.of(12000000L, 17999999L)))
+        );
+    }
+
 }
\ No newline at end of file
diff --git a/services-custom/s3-transfer-manager/src/test/resources/log4j2.properties b/services-custom/s3-transfer-manager/src/test/resources/log4j2.properties
index 85978ec46781..6296554b588a 100644
--- a/services-custom/s3-transfer-manager/src/test/resources/log4j2.properties
+++ b/services-custom/s3-transfer-manager/src/test/resources/log4j2.properties
@@ -24,10 +24,10 @@ rootLogger.level = info
 rootLogger.appenderRef.stdout.ref = ConsoleAppender
 
 logger.split.name = software.amazon.awssdk.core.internal.async
-logger.split.level = trace
+logger.split.level = debug
 
 logger.multi.name = software.amazon.awssdk.services.s3.internal.multipart
-logger.multi.level = trace
+logger.multi.level = debug
 
 # Uncomment below to enable more specific logging
 #
diff --git a/services/s3/src/it/java/software/amazon/awssdk/services/s3/multipart/S3MultipartClientFileDownloadIntegrationTest.java b/services/s3/src/it/java/software/amazon/awssdk/services/s3/multipart/S3MultipartClientFileDownloadIntegrationTest.java
new file mode 100644
index 000000000000..a9fced5873f2
--- /dev/null
+++ b/services/s3/src/it/java/software/amazon/awssdk/services/s3/multipart/S3MultipartClientFileDownloadIntegrationTest.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.services.s3.multipart;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.security.MessageDigest;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import software.amazon.awssdk.core.FileTransformerConfiguration;
+import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.async.AsyncRequestBody;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.core.interceptor.Context;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3IntegrationTestBase;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.testutils.RandomTempFile;
+import software.amazon.awssdk.utils.JavaSystemSetting;
+import software.amazon.awssdk.utils.Logger;
+
+@Timeout(value = 5, unit = TimeUnit.MINUTES)
+public class S3MultipartClientFileDownloadIntegrationTest extends S3IntegrationTestBase {
+    private static final Logger log = Logger.loggerFor(S3MultipartClientFileDownloadIntegrationTest.class);
+    private static final int MIB = 1024 * 1024;
+    private static final String TEST_BUCKET = temporaryBucketName(S3MultipartClientFileDownloadIntegrationTest.class);
+    private static final String TEST_KEY = "testfile.dat";
+    private static final int OBJ_SIZE = 100 * MIB;
+    private static final long PART_SIZE = 5 * MIB;
+
+    private static RandomTempFile localFile;
+    private S3AsyncClient s3Client;
+    private TestInterceptor interceptor;
+
+    @BeforeAll
+    public static void setup() throws Exception {
+        log.info(() -> "setup");
+        setUp();
+        log.info(() -> "create bucket");
+        createBucket(TEST_BUCKET);
+        localFile = new RandomTempFile(TEST_KEY, OBJ_SIZE);
+        localFile.deleteOnExit();
+        S3AsyncClient s3Client = S3AsyncClient.builder()
+                                              .multipartEnabled(true)
+                                              .multipartConfiguration(c -> c.minimumPartSizeInBytes(PART_SIZE))
+                                              .credentialsProvider(CREDENTIALS_PROVIDER_CHAIN)
+                                              .build();
+        log.info(() -> "put multipart object");
+        s3Client.putObject(r -> r.bucket(TEST_BUCKET).key(TEST_KEY), AsyncRequestBody.fromFile(localFile))
+                .join();
+        s3.close();
+    }
+
+    @BeforeEach
+    public void init() {
+        log.info(() -> "Initializing S3MultipartClientFileDownloadIntegrationTest");
+        this.interceptor = new TestInterceptor();
+        this.s3Client = S3AsyncClient.builder()
+                                     .multipartEnabled(true)
+                                     .multipartConfiguration(c -> c.maxInflightDownloads(50))
+                                     .overrideConfiguration(o -> o.addExecutionInterceptor(this.interceptor))
+                                     .credentialsProvider(CREDENTIALS_PROVIDER_CHAIN)
+                                     .build();
+    }
+
+    @Test
+    void download_defaultCreateNewFile_shouldSucceed() throws Exception {
+        Path path = tmpPath().resolve(UUID.randomUUID().toString());
+        CompletableFuture<GetObjectResponse> future = s3Client.getObject(
+            req -> req.bucket(TEST_BUCKET).key(TEST_KEY),
+            AsyncResponseTransformer.toFile(path, FileTransformerConfiguration.defaultCreateNew()));
+        future.join();
+        assertSameContentWithChecksum(path);
+        int totalParts = OBJ_SIZE / (int) PART_SIZE;
+        assertThat(interceptor.parts.size()).isEqualTo(totalParts);
+        assertThat(interceptor.parts).hasSameElementsAs(IntStream.range(1, totalParts +1).boxed().collect(Collectors.toList()));
+        path.toFile().delete();
+    }
+
+    private Path tmpPath() {
+        return Paths.get(JavaSystemSetting.TEMP_DIRECTORY.getStringValueOrThrow());
+    }
+
+    private static final class TestInterceptor implements ExecutionInterceptor {
+        List<Integer> parts = new ArrayList<>();
+
+        @Override
+        public void beforeTransmission(Context.BeforeTransmission context, ExecutionAttributes executionAttributes) {
+            SdkRequest request = context.request();
+            if (request instanceof GetObjectRequest) {
+                log.info(() -> "Received GetObjectRequest for request " + request);
+                GetObjectRequest getObjectRequest = (GetObjectRequest) request;
+                parts.add(getObjectRequest.partNumber());
+            } else {
+                log.warn(() -> "Unexpected request type: " + request.getClass());
+            }
+        }
+    }
+
+    private void assertSameContentWithChecksum(Path path) throws Exception {
+        MessageDigest md = MessageDigest.getInstance("SHA-256");
+        byte[] downloadedHash = md.digest(Files.readAllBytes(path));
+        md.reset();
+        byte[] originalHash = md.digest(Files.readAllBytes(localFile.toPath()));
+        assertThat(downloadedHash).isEqualTo(originalHash);
+    }
+}
\ No newline at end of file
diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/DownloadObjectHelper.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/DownloadObjectHelper.java
index 2d6fadc5f505..6cd19da3de1a 100644
--- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/DownloadObjectHelper.java
+++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/DownloadObjectHelper.java
@@ -31,10 +31,12 @@ public class DownloadObjectHelper {
 
     private final S3AsyncClient s3AsyncClient;
     private final long bufferSizeInBytes;
+    private final int maxInFlight;
 
-    public DownloadObjectHelper(S3AsyncClient s3AsyncClient, long bufferSizeInBytes) {
+    public DownloadObjectHelper(S3AsyncClient s3AsyncClient, long bufferSizeInBytes, int maxInFlight) {
         this.s3AsyncClient = s3AsyncClient;
         this.bufferSizeInBytes = bufferSizeInBytes;
+        this.maxInFlight = maxInFlight;
     }
 
     public <T> CompletableFuture<T> downloadObject(
@@ -47,6 +49,27 @@ public <T> CompletableFuture<T> downloadObject(
             asyncResponseTransformer.split(SplittingTransformerConfiguration.builder()
                                                                             .bufferSizeInBytes(bufferSizeInBytes)
                                                                             .build());
+        if (!split.supportsNonSerial()) {
+            return downloadPartsSerially(getObjectRequest, split);
+        }
+
+        return downloadPartsNonSerially(getObjectRequest, split, maxInFlight);
+
+    }
+
+    private <T> CompletableFuture<T> downloadPartsNonSerially(
+        GetObjectRequest getObjectRequest,
+        AsyncResponseTransformer.SplitResult<GetObjectResponse, T> split,
+        int maxInFlight) {
+        // TODO pause & resume
+        NonLinearMultipartDownloaderSubscriber subscriber = new NonLinearMultipartDownloaderSubscriber(
+            s3AsyncClient, getObjectRequest, (CompletableFuture<GetObjectResponse>) split.resultFuture(), maxInFlight);
+        split.publisher().subscribe(subscriber);
+        return split.resultFuture();
+    }
+
+    private <T> CompletableFuture<T> downloadPartsSerially(GetObjectRequest getObjectRequest,
+                                                           AsyncResponseTransformer.SplitResult<GetObjectResponse, T> split) {
         MultipartDownloaderSubscriber subscriber = subscriber(getObjectRequest);
         split.publisher().subscribe(subscriber);
         return split.resultFuture();
diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartConfigurationResolver.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartConfigurationResolver.java
index 9fc199175bda..c63a66bb672e 100644
--- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartConfigurationResolver.java
+++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartConfigurationResolver.java
@@ -26,9 +26,11 @@
 public final class MultipartConfigurationResolver {
 
     private static final long DEFAULT_MIN_PART_SIZE = 8L * 1024 * 1024;
+    private static final int DEFAULT_MAX_IN_FLIGHT = 50;
     private final long minimalPartSizeInBytes;
     private final long apiCallBufferSize;
     private final long thresholdInBytes;
+    private final int maxInFlight;
 
     public MultipartConfigurationResolver(MultipartConfiguration multipartConfiguration) {
         Validate.notNull(multipartConfiguration, "multipartConfiguration");
@@ -37,6 +39,7 @@ public MultipartConfigurationResolver(MultipartConfiguration multipartConfigurat
         this.apiCallBufferSize = Validate.getOrDefault(multipartConfiguration.apiCallBufferSizeInBytes(),
                                                        () -> minimalPartSizeInBytes * 4);
         this.thresholdInBytes = Validate.getOrDefault(multipartConfiguration.thresholdInBytes(), () -> minimalPartSizeInBytes);
+        this.maxInFlight = Validate.getOrDefault(multipartConfiguration.maxInflightDownloads(), () -> DEFAULT_MAX_IN_FLIGHT);
     }
 
     public long minimalPartSizeInBytes() {
@@ -50,4 +53,8 @@ public long thresholdInBytes() {
     public long apiCallBufferSize() {
         return apiCallBufferSize;
     }
+
+    public int maxInFlight() {
+        return maxInFlight;
+    }
 }
diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartS3AsyncClient.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartS3AsyncClient.java
index 6142db7772f7..3ba362ccae55 100644
--- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartS3AsyncClient.java
+++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartS3AsyncClient.java
@@ -63,9 +63,10 @@ private MultipartS3AsyncClient(S3AsyncClient delegate, MultipartConfiguration mu
         long minPartSizeInBytes = resolver.minimalPartSizeInBytes();
         long threshold = resolver.thresholdInBytes();
         long apiCallBufferSize = resolver.apiCallBufferSize();
+        int maxInFLight = resolver.maxInFlight();
         mpuHelper = new UploadObjectHelper(delegate, resolver);
         copyObjectHelper = new CopyObjectHelper(delegate, minPartSizeInBytes, threshold);
-        downloadObjectHelper = new DownloadObjectHelper(delegate, apiCallBufferSize);
+        downloadObjectHelper = new DownloadObjectHelper(delegate, apiCallBufferSize, maxInFLight);
         this.checksumEnabled = checksumEnabled;
     }
 
diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartUploadHelper.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartUploadHelper.java
index e752a40e6262..2fc8b58a00f2 100644
--- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartUploadHelper.java
+++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartUploadHelper.java
@@ -78,10 +78,10 @@ CompletableFuture<CreateMultipartUploadResponse> createMultipartUpload(PutObject
     }
 
     CompletableFuture<CompleteMultipartUploadResponse> completeMultipartUpload(CompletableFuture<PutObjectResponse> returnFuture,
-                                 String uploadId,
-                                 CompletedPart[] completedParts,
-                                 PutObjectRequest putObjectRequest,
-                                 long contentLength) {
+                                                                               String uploadId,
+                                                                               CompletedPart[] completedParts,
+                                                                               PutObjectRequest putObjectRequest,
+                                                                               long contentLength) {
         CompletableFuture<CompleteMultipartUploadResponse> future =
             genericMultipartHelper.completeMultipartUpload(putObjectRequest, uploadId, completedParts, contentLength);
 
@@ -102,14 +102,14 @@ CompletableFuture<CompletedPart> sendIndividualUploadPartRequest(String uploadId
         UploadPartRequest uploadPartRequest = requestPair.left();
         Integer partNumber = uploadPartRequest.partNumber();
         Optional<Long> contentLength = requestPair.right().contentLength();
-        log.debug(() -> "Sending uploadPartRequest: " + uploadPartRequest.partNumber() + " uploadId: " + uploadId + " "
-                        + "contentLength " + contentLength);
+        log.info(() -> "Sending uploadPartRequest: " + uploadPartRequest.partNumber());
 
         CompletableFuture<UploadPartResponse> uploadPartFuture = s3AsyncClient.uploadPart(uploadPartRequest,
                                                                                           requestPair.right());
 
         CompletableFuture<CompletedPart> convertFuture =
             uploadPartFuture.thenApply(uploadPartResponse -> {
+                log.info(() -> "Completed part: " + partNumber);
                 contentLength.ifPresent(progressListener::subscriberOnNext);
                 return convertUploadPartResponse(completedPartsConsumer, partNumber, uploadPartResponse);
             });
diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java
new file mode 100644
index 000000000000..d0b30ef4f7e6
--- /dev/null
+++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java
@@ -0,0 +1,395 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.services.s3.internal.multipart;
+
+import java.util.ArrayDeque;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.utils.CompletableFutureUtils;
+import software.amazon.awssdk.utils.Logger;
+
+@SdkInternalApi
+public class NonLinearMultipartDownloaderSubscriber
+    implements Subscriber<AsyncResponseTransformer<GetObjectResponse, GetObjectResponse>> {
+    private static final Logger log = Logger.loggerFor(NonLinearMultipartDownloaderSubscriber.class);
+
+    /**
+     * Maximum number of concurrent GetObject requests
+     */
+    private final int maxInFlight;
+
+    /**
+     * The s3 client used to make the individual part requests
+     */
+    private final S3AsyncClient s3;
+
+    /**
+     * The GetObjectRequest that was provided when calling s3.getObject(...). It is copied for each individual request, and the
+     * copy has the partNumber field updated as more parts are downloaded.
+     */
+    private final GetObjectRequest getObjectRequest;
+
+    /**
+     * The total number of completed parts. A part is considered complete once the completable future associated with its request
+     * completes successfully.
+     */
+    private final AtomicInteger completedParts = new AtomicInteger();
+
+    /**
+     * Queue of all remaining parts in ascending order. i.e. 4, 5, 6, 8, 11
+     */
+    private final Queue<Integer> allRemainingParts = new ArrayDeque<>();
+
+    /**
+     * The future returned to the user when calling
+     * {@link S3AsyncClient#getObject(GetObjectRequest, AsyncResponseTransformer) getObject}. This will be completed once the last
+     * part finishes. Contrary to the linear code path, the future returned to the user is handled here so that we can complete it
+     * once the last part writting to the file is completed.
+     */
+    private final CompletableFuture<GetObjectResponse> resultFuture;
+
+    /**
+     * The {@link GetObjectResponse} to be returned in the completed future to the user. It corresponds to the response of first
+     * part GetObject
+     */
+    private GetObjectResponse getObjectResponse;
+
+    /**
+     * The subscription received from the publisher this subscriber subscribes to.
+     */
+    private Subscription subscription;
+
+    /**
+     * This value indicates the total number of parts of the object to get. If null, it means we don't know the total amount of
+     * parts, either because we haven't received a response from s3 yet to set it, or the object to get is not multipart.
+     */
+    private volatile Integer totalParts;
+
+    /**
+     * The etag of the object being downloaded.
+     */
+    private volatile String eTag;
+
+    /**
+     * Tracks request that are currently in flights, waiting to be completed. Once completed, future are removed from the map
+     */
+    private final Map<Integer, CompletableFuture<GetObjectResponse>> inFlightRequests = new ConcurrentHashMap<>();
+
+    /**
+     * Lock used during first part, so that the first batch of requests wait to be executed until we know if the s3 object is
+     * multipart or not.
+     */
+    private final ReentrantLock firstPartLock = new ReentrantLock();
+
+    /**
+     * Pending transformers waiting for the first request to be executed. We cant execute multiple part get until the first
+     * requests completes, and we know the object in s3 is a multipart object.
+     */
+    private final Queue<AsyncResponseTransformer<GetObjectResponse, GetObjectResponse>> pendingTransformers = new ArrayDeque<>();
+
+    /**
+     * Tracks if the first request was sent
+     */
+    private final AtomicBoolean firstRequestSent = new AtomicBoolean(false);
+
+    /**
+     * Tracks the total number of transformers requested from the subscription
+     */
+    private final AtomicInteger transformersRequested = new AtomicInteger(0);
+
+    /**
+     * Amount of demand requested but not yet fulfilled by the subscription
+     */
+    private final AtomicInteger outstandingDemand = new AtomicInteger(0);
+
+    public NonLinearMultipartDownloaderSubscriber(S3AsyncClient s3,
+                                                  GetObjectRequest getObjectRequest,
+                                                  CompletableFuture<GetObjectResponse> resultFuture,
+                                                  int maxInFLight) {
+        this.s3 = s3;
+        this.getObjectRequest = getObjectRequest;
+        this.resultFuture = resultFuture;
+        this.maxInFlight = maxInFLight;
+    }
+
+    @Override
+    public void onSubscribe(Subscription s) {
+        if (this.subscription != null) {
+            s.cancel();
+            return;
+        }
+        this.subscription = s;
+        request(1);
+    }
+
+    private void request(int amount) {
+        outstandingDemand.addAndGet(amount);
+        transformersRequested.addAndGet(amount);
+        subscription.request(amount);
+    }
+
+    @Override
+    public void onNext(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
+        // todo change to trace log
+        outstandingDemand.decrementAndGet();
+        log.trace(() -> "=== On Next ===\nTotal in flight parts: " + inFlightRequests.size()
+                       + "\nOutstanding Demand : " + outstandingDemand.get()
+                       + "\nTotal completed parts: " + completedParts
+                       + "\nTotal transformers requested: " + transformersRequested.get()
+                       + "\nTotal pending transformers: " + pendingTransformers.size()
+                       + "\nCurrent in flight requests: " + inFlightRequests.keySet());
+        if (asyncResponseTransformer == null) {
+            subscription.cancel();
+            throw new NullPointerException("onNext must not be called with null asyncResponseTransformer");
+        }
+        executeRequestOrAddToPending(asyncResponseTransformer);
+    }
+
+    private void executeRequestOrAddToPending(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
+        if (handleFirstRequest(asyncResponseTransformer)) {
+            return;
+        }
+
+        if (inFlightRequests.size() >= maxInFlight) {
+            pendingTransformers.offer(asyncResponseTransformer);
+            return;
+        }
+
+        sendNextRequest(asyncResponseTransformer);
+        requestMoreIfNeeded();
+    }
+
+    // Handle first request.
+    // We need to wait for the first request to finish so we know if it is aa multipart object or not.
+    // While we don't know yet, additional onNext signal receives are stored in pendingTransformers.
+    private boolean handleFirstRequest(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
+        if (completedParts.get() != 0) {
+            return false;
+        }
+
+        firstPartLock.lock();
+        try {
+            if (!firstRequestSent.get()) {
+                sendFirstRequest(asyncResponseTransformer);
+                firstRequestSent.set(true);
+                return true;
+            }
+
+            // First request already sent, queue this transformer
+            synchronized (pendingTransformers) {
+                pendingTransformers.offer(asyncResponseTransformer);
+            }
+        } finally {
+            firstPartLock.unlock();
+        }
+        return true;
+    }
+
+    private void sendNextRequest(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
+        if (inFlightRequests.size() + completedParts.get() >= totalParts) {
+            return;
+        }
+
+        Integer partToGet = nextPart();
+
+        GetObjectRequest request = nextRequest(partToGet);
+        // todo change to trace log
+        log.trace(() -> "Sending next request for part: " + partToGet);
+
+        CompletableFuture<GetObjectResponse> response = s3.getObject(request, asyncResponseTransformer);
+
+        inFlightRequests.put(partToGet, response);
+        CompletableFutureUtils.forwardExceptionTo(resultFuture, response);
+
+        response.whenComplete((res, e) -> {
+            log.trace(() -> "Completed part: " + partToGet);
+            inFlightRequests.remove(partToGet);
+
+            completedParts.incrementAndGet();
+            if (e != null) {
+                // Note on retries: When this future completes exceptionally, it means we did all retries and still failed for
+                // that part. We need to report back the failure to the user.
+                log.error(() -> "Error on part " + partToGet + ": " + e);
+                resultFuture.completeExceptionally(e);
+                inFlightRequests.values().forEach(future -> future.cancel(true));
+                return;
+            }
+            if (completedParts.get() >= totalParts) {
+                // todo remove log
+                log.info(() -> "================= ALL PARTS COMPLETED ==================");
+                subscription.cancel();
+                resultFuture.complete(getObjectResponse);
+            } else {
+                processPendingTransformers();
+                requestMoreIfNeeded();
+            }
+        });
+    }
+
+    // returns true if the first request was sent and is still in flight
+    private void sendFirstRequest(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
+        log.debug(() -> "Sending first request");
+        GetObjectRequest request = nextRequest(1);
+        CompletableFuture<GetObjectResponse> responseFuture = s3.getObject(request, asyncResponseTransformer);
+        inFlightRequests.put(1, responseFuture);
+
+        // Propagate cancellation from user
+        CompletableFutureUtils.forwardExceptionTo(resultFuture, responseFuture);
+
+        responseFuture.whenComplete((res, e) -> {
+           inFlightRequests.remove(1);
+
+            completedParts.incrementAndGet();
+            if (e != null) {
+                // Note on retries: When this future completes exceptionally, it means we did all retries and still failed for
+                // that part. We need to report back the failure to the user.
+                resultFuture.completeExceptionally(e);
+                inFlightRequests.values().forEach(future -> future.cancel(true));
+                return;
+            }
+            Integer partCount = res.partsCount();
+            eTag = res.eTag();
+            if (partCount != null && totalParts == null) {
+                // todo change debug log
+                log.debug(() -> String.format("Total amount of parts of the object to download: %d", partCount));
+                // MultipartDownloadUtils.multipartDownloadResumeContext(getObjectRequest)
+                //                       .ifPresent(ctx -> ctx.totalParts(partCount));
+                totalParts = partCount;
+            }
+
+            // it is a multipart object?
+            if (totalParts == null || totalParts == 1) {
+                // Single part object detected, skip multipart and complete everything now
+                // todo change debug log
+                log.debug(() -> "Single Part object detected, skipping multipart download");
+                subscription.cancel();
+                resultFuture.complete(res);
+                return;
+            }
+
+            // todo change debug log
+            log.debug(() -> "Multipart object detected, performing multipart download");
+            // part 1 already completed, so skip it
+            for (int i = 2; i <= totalParts; i++) {
+                allRemainingParts.add(i);
+            }
+            getObjectResponse = res;
+
+            processPendingTransformers();
+            requestMoreIfNeeded();
+        });
+    }
+
+    private void processPendingTransformers() {
+        synchronized (pendingTransformers) {
+            int initialSize = pendingTransformers.size();
+            for (int i = 0; i < initialSize; i++) {
+                AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> transformer = pendingTransformers.poll();
+                if (transformer != null) {
+                    executeRequestOrAddToPending(transformer);
+                }
+            }
+        }
+    }
+
+    private void requestMoreIfNeeded() {
+        if (totalParts == null) {
+            return; // Don't know total parts yet
+        }
+
+        int currentRequested = transformersRequested.get();
+        if (currentRequested >= totalParts) {
+            return; // Already requested enough transformers
+        }
+
+        // Don't request more if we already have enough work in progress
+        int totalWorkInProgress = inFlightRequests.size() + completedParts.get();
+        if (totalWorkInProgress >= totalParts) {
+            log.trace(() -> "Not requesting more. Total work in progress (" + totalWorkInProgress
+                            + ") >= totalParts (" + totalParts + ")");
+            return;
+        }
+
+        // Only request more if we have capacity and remaining parts
+        int remainingParts = allRemainingParts.size();
+        if (remainingParts == 0) {
+            return;
+        }
+
+        // don't request if we already have enough work in progress
+        if (inFlightRequests.size() >= maxInFlight) {
+            return;
+        }
+
+        // don't request if we have already requested more work than we can handle
+        if (outstandingDemand.get() >= maxInFlight) {
+            return;
+        }
+
+        int partsNeeded = Math.min(Math.min(totalParts - currentRequested, remainingParts),
+                                   maxInFlight - inFlightRequests.size());
+        if (partsNeeded > 0) {
+            log.trace(() -> "Requesting " + partsNeeded + " more transformers. Total requested will be: "
+                            + (currentRequested + partsNeeded) + ". Remaining parts: " + remainingParts
+                            + ". InFlight: " + inFlightRequests.size());
+            request(partsNeeded);
+        }
+
+    }
+
+    private Integer nextPart() {
+        synchronized (allRemainingParts) {
+            return allRemainingParts.poll();
+        }
+    }
+
+    @Override
+    public void onError(Throwable t) {
+        // Signal received from the publisher this is subscribed to, in the case of file download, that's
+        // FileAsyncResponseTransformerPublisher.
+        // failed state, something really wrong has happened, cancel everything
+        inFlightRequests.values().forEach(future -> future.cancel(true));
+        inFlightRequests.clear();
+        resultFuture.completeExceptionally(t);
+    }
+
+    @Override
+    public void onComplete() {
+
+    }
+
+    private GetObjectRequest nextRequest(int nextPartToGet) {
+        return getObjectRequest.copy(req -> {
+            req.partNumber(nextPartToGet);
+            if (eTag != null) {
+                req.ifMatch(eTag);
+            }
+        });
+    }
+
+}
diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/UploadWithKnownContentLengthHelper.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/UploadWithKnownContentLengthHelper.java
index cd4ebc4a88b9..52729856d358 100644
--- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/UploadWithKnownContentLengthHelper.java
+++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/UploadWithKnownContentLengthHelper.java
@@ -182,7 +182,9 @@ private void splitAndSubscribe(MpuRequestContext mpuRequestContext, CompletableF
             new KnownContentLengthAsyncRequestBodySubscriber(mpuRequestContext, returnFuture, multipartUploadHelper);
 
         attachSubscriberToObservable(subscriber, mpuRequestContext.request().left());
+        log.debug(() -> "splitting async request body with part size: " + mpuRequestContext.partSize());
 
+        log.info(() -> "Split: chunkSizeInBytes: " + mpuRequestContext.partSize() + " - bufferSizeInBytes: " + maxMemoryUsageInBytes);
         mpuRequestContext.request().right()
             .split(b -> b.chunkSizeInBytes(mpuRequestContext.partSize())
                          .bufferSizeInBytes(maxMemoryUsageInBytes))
diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/multipart/MultipartConfiguration.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/multipart/MultipartConfiguration.java
index 9b074d6244f3..d586afdaa8a2 100644
--- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/multipart/MultipartConfiguration.java
+++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/multipart/MultipartConfiguration.java
@@ -44,11 +44,13 @@ public final class MultipartConfiguration implements ToCopyableBuilder<Multipart
     private final Long thresholdInBytes;
     private final Long minimumPartSizeInBytes;
     private final Long apiCallBufferSizeInBytes;
+    private final Integer maxInflightDownloads;
 
     private MultipartConfiguration(DefaultMultipartConfigBuilder builder) {
         this.thresholdInBytes = builder.thresholdInBytes;
         this.minimumPartSizeInBytes = builder.minimumPartSizeInBytes;
         this.apiCallBufferSizeInBytes = builder.apiCallBufferSizeInBytes;
+        this.maxInflightDownloads = builder.maxInflightDownloads;
     }
 
     public static Builder builder() {
@@ -89,6 +91,10 @@ public Long apiCallBufferSizeInBytes() {
         return this.apiCallBufferSizeInBytes;
     }
 
+    public Integer maxInflightDownloads() {
+        return this.maxInflightDownloads;
+    }
+
     /**
      * Builder for a {@link MultipartConfiguration}.
      */
@@ -163,12 +169,27 @@ public interface Builder extends CopyableBuilder<Builder, MultipartConfiguration
          * @return the value of the maximum memory usage.
          */
         Long apiCallBufferSizeInBytes();
+
+        /**
+         * todo
+         * @param maxInflightDownloads
+         * @return
+         */
+        Builder maxInflightDownloads(Integer maxInflightDownloads);
+
+        /**
+         * todo
+         * @return
+         */
+        Integer maxInflightDownloads();
+
     }
 
     private static class DefaultMultipartConfigBuilder implements Builder {
         private Long thresholdInBytes;
         private Long minimumPartSizeInBytes;
         private Long apiCallBufferSizeInBytes;
+        private Integer maxInflightDownloads;
 
         public Builder thresholdInBytes(Long thresholdInBytes) {
             this.thresholdInBytes = thresholdInBytes;
@@ -188,6 +209,17 @@ public Long minimumPartSizeInBytes() {
             return this.minimumPartSizeInBytes;
         }
 
+        @Override
+        public Builder maxInflightDownloads(Integer maxInflightDownloads) {
+            this.maxInflightDownloads = maxInflightDownloads;
+            return this;
+        }
+
+        @Override
+        public Integer maxInflightDownloads() {
+            return this.maxInflightDownloads;
+        }
+
         @Override
         public Builder apiCallBufferSizeInBytes(Long maximumMemoryUsageInBytes) {
             this.apiCallBufferSizeInBytes = maximumMemoryUsageInBytes;
diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartDownloadTestUtil.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartDownloadTestUtil.java
index 708972b6b0d7..9523cfdf1c8f 100644
--- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartDownloadTestUtil.java
+++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartDownloadTestUtil.java
@@ -63,17 +63,25 @@ public byte[] stubAllParts(String testBucket, String testKey, int amountOfPartTo
         return expectedBody;
     }
 
-    public byte[] stubForPart(String testBucket, String testKey,int part, int totalPart, int partSize) {
+    public byte[] stubForPart(String testBucket, String testKey, int part, int totalPart, int partSize) {
         byte[] body = new byte[partSize];
         random.nextBytes(body);
         stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=%d", testBucket, testKey, part))).willReturn(
             aResponse()
                 .withHeader("x-amz-mp-parts-count", totalPart + "")
+                .withHeader("x-amz-content-range", contentRangeHeader(part, totalPart, partSize))
                 .withHeader("ETag", eTag)
                 .withBody(body)));
         return body;
     }
 
+    public static String contentRangeHeader(int part, int totalPart, int partSize) {
+        long start = (part - 1) * (long) partSize;
+        long end = start + partSize - 1;
+        long total = totalPart * (long) partSize;
+        return String.format("bytes %d-%d/%d", start, end, total);
+    }
+
     public void verifyCorrectAmountOfRequestsMade(int amountOfPartToTest) {
         String urlTemplate = ".*partNumber=%d.*";
         for (int i = 1; i <= amountOfPartToTest; i++) {
@@ -95,4 +103,16 @@ public byte[] stubForPartSuccess(int part, int totalPart, int partSize) {
                             .withBody(body)));
         return body;
     }
+
+    public byte[] stubSinglePart(String testBucket, String testKey, int partSize) {
+        byte[] body = new byte[partSize];
+        random.nextBytes(body);
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey))).willReturn(
+            aResponse()
+                .withHeader("x-amz-mp-parts-count", "1")
+                .withHeader("x-amz-content-range", String.format("bytes %d-%d/%d", 0, partSize - 1, partSize))
+                .withHeader("ETag", eTag)
+                .withBody(body)));
+        return body;
+    }
 }
diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriberWiremockTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriberWiremockTest.java
new file mode 100644
index 000000000000..d44d40011987
--- /dev/null
+++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriberWiremockTest.java
@@ -0,0 +1,140 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.services.s3.internal.multipart;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
+import com.github.tomakehurst.wiremock.junit5.WireMockTest;
+import com.google.common.jimfs.Configuration;
+import com.google.common.jimfs.Jimfs;
+import java.net.URI;
+import java.nio.file.FileSystem;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.CompletableFuture;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.reactivestreams.Subscriber;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.SplittingTransformerConfiguration;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+
+@WireMockTest
+class NonLinearMultipartDownloaderSubscriberWiremockTest {
+
+    private final String testBucket = "test-bucket";
+    private final String testKey = "test-key";
+
+    private S3AsyncClient s3AsyncClient;
+    private MultipartDownloadTestUtil util;
+    private FileSystem fileSystem;
+    private Path testFile;
+
+    @BeforeEach
+    public void init(WireMockRuntimeInfo wiremock) throws Exception {
+        s3AsyncClient = S3AsyncClient.builder()
+                                     .credentialsProvider(StaticCredentialsProvider.create(
+                                         AwsBasicCredentials.create("key", "secret")))
+                                     .region(Region.US_WEST_2)
+                                     .endpointOverride(URI.create("http://localhost/" + wiremock.getHttpPort()))
+                                     .serviceConfiguration(S3Configuration.builder()
+                                                                          .pathStyleAccessEnabled(true)
+                                                                          .build())
+                                     .build();
+        util = new MultipartDownloadTestUtil(testBucket, testKey, "test-etag");
+        fileSystem = Jimfs.newFileSystem(Configuration.unix());
+        testFile = fileSystem.getPath("/test-file.txt");
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        fileSystem.close();
+    }
+
+    @ParameterizedTest
+    @ValueSource(ints = {2, 3, 4, 5, 6, 7, 8, 9, 10})
+    void happyPath_multipartDownload(int numParts) throws Exception {
+        int partSize = 1024;
+        byte[] expectedBody = util.stubAllParts(testBucket, testKey, numParts, partSize);
+
+        AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> transformer = AsyncResponseTransformer.toFile(testFile);
+        AsyncResponseTransformer.SplitResult<GetObjectResponse, GetObjectResponse> split = transformer.split(
+            SplittingTransformerConfiguration.builder()
+                                             .bufferSizeInBytes(1024 * 32L)
+                                             .build());
+
+        CompletableFuture<GetObjectResponse> resultFuture = new CompletableFuture<>();
+        Subscriber<AsyncResponseTransformer<GetObjectResponse, GetObjectResponse>> subscriber =
+            new NonLinearMultipartDownloaderSubscriber(s3AsyncClient,
+                                                       GetObjectRequest.builder()
+                                                                       .bucket(testBucket)
+                                                                       .key(testKey)
+                                                                       .build(),
+                                                       resultFuture,
+                                                       50);
+
+        split.publisher().subscribe(subscriber);
+        GetObjectResponse getObjectResponse = resultFuture.join();
+
+        assertThat(Files.exists(testFile)).isTrue();
+        byte[] actualBody = Files.readAllBytes(testFile);
+        assertThat(actualBody).isEqualTo(expectedBody);
+        assertThat(getObjectResponse).isNotNull();
+        util.verifyCorrectAmountOfRequestsMade(numParts);
+    }
+
+    @Test
+    void singlePartObject_shouldCompleteWithoutMultipart() throws Exception {
+        int partSize = 1024;
+        byte[] expectedBody = util.stubSinglePart(testBucket, testKey, partSize);
+
+        AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> transformer = AsyncResponseTransformer.toFile(testFile);
+        AsyncResponseTransformer.SplitResult<GetObjectResponse, GetObjectResponse> split = transformer.split(
+            SplittingTransformerConfiguration.builder()
+                                             .bufferSizeInBytes(1024 * 32L)
+                                             .build());
+
+        CompletableFuture<GetObjectResponse> resultFuture = new CompletableFuture<>();
+        Subscriber<AsyncResponseTransformer<GetObjectResponse, GetObjectResponse>> subscriber =
+            new NonLinearMultipartDownloaderSubscriber(s3AsyncClient,
+                                                       GetObjectRequest.builder()
+                                                                       .bucket(testBucket)
+                                                                       .key(testKey)
+                                                                       .build(),
+                                                       resultFuture,
+                                                       50);
+
+        split.publisher().subscribe(subscriber);
+        GetObjectResponse getObjectResponse = resultFuture.join();
+
+        assertThat(Files.exists(testFile)).isTrue();
+        byte[] actualBody = Files.readAllBytes(testFile);
+        assertThat(actualBody).isEqualTo(expectedBody);
+        assertThat(getObjectResponse).isNotNull();
+        util.verifyCorrectAmountOfRequestsMade(1);
+    }
+
+}
\ No newline at end of file
diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/S3MultipartFileDownloadWiremockTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/S3MultipartFileDownloadWiremockTest.java
new file mode 100644
index 000000000000..6a5199e683e3
--- /dev/null
+++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/S3MultipartFileDownloadWiremockTest.java
@@ -0,0 +1,425 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.services.s3.internal.multipart;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.exactly;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
+import static org.assertj.core.api.Assertions.assertThat;
+import static software.amazon.awssdk.services.s3.internal.multipart.MultipartDownloadTestUtil.contentRangeHeader;
+
+import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
+import com.github.tomakehurst.wiremock.junit5.WireMockTest;
+import com.google.common.jimfs.Configuration;
+import com.google.common.jimfs.Jimfs;
+import java.net.URI;
+import java.nio.file.FileSystem;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+@WireMockTest
+class S3MultipartFileDownloadWiremockTest {
+
+    private final String testBucket = "test-bucket";
+    private final String testKey = "test-key";
+
+    private S3AsyncClient s3AsyncClient;
+    private MultipartDownloadTestUtil util;
+    private FileSystem fileSystem;
+    private Path testFile;
+
+    @BeforeEach
+    public void init(WireMockRuntimeInfo wiremock) throws Exception {
+        s3AsyncClient = S3AsyncClient.builder()
+                                     .credentialsProvider(StaticCredentialsProvider.create(
+                                         AwsBasicCredentials.create("key", "secret")))
+                                     .region(Region.US_WEST_2)
+                                     .endpointOverride(URI.create("http://localhost/" + wiremock.getHttpPort()))
+                                     .multipartEnabled(true)
+                                     .serviceConfiguration(S3Configuration.builder()
+                                                                          .pathStyleAccessEnabled(true)
+                                                                          .build())
+                                     .httpClient(NettyNioAsyncHttpClient.builder()
+                                                                        // .maxConcurrency(10_000)
+                                                                        // .connectionAcquisitionTimeout(Duration.ofSeconds(5))
+                                                                        // .connectionTimeout(Duration.ofSeconds(5))
+                                                                        // .connectionTimeToLive(Duration.ofSeconds(5))
+                                                                        .build())
+                                     // .httpClient(AwsCrtAsyncHttpClient.create())
+                                     .build();
+        util = new MultipartDownloadTestUtil(testBucket, testKey, "test-etag");
+        fileSystem = Jimfs.newFileSystem(Configuration.unix());
+        testFile = fileSystem.getPath("/test-file.txt");
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        fileSystem.close();
+    }
+
+    @Test
+    void happyPath_singlePart() throws Exception {
+        int partSize = 1024;
+        byte[] expectedBody = util.stubSinglePart(testBucket, testKey, partSize);
+
+        CompletableFuture<GetObjectResponse> response = s3AsyncClient.getObject(b -> b
+                                                                                    .bucket(testBucket)
+                                                                                    .key(testKey)
+                                                                                    .build(),
+                                                                                AsyncResponseTransformer.toFile(testFile));
+
+        assertThat(response).succeedsWithin(Duration.of(10, ChronoUnit.SECONDS));
+        assertThat(Files.exists(testFile)).isTrue();
+        byte[] actualBody = Files.readAllBytes(testFile);
+        assertThat(actualBody).isEqualTo(expectedBody);
+        assertThat(response).isNotNull();
+        util.verifyCorrectAmountOfRequestsMade(1);
+    }
+
+    @Test
+    void happyPath_multipart() throws Exception {
+        int numParts = 4;
+        int partSize = 1024;
+        byte[] expectedBody = util.stubAllParts(testBucket, testKey, numParts, partSize);
+
+        CompletableFuture<GetObjectResponse> response = s3AsyncClient.getObject(b -> b
+                                                                                    .bucket(testBucket)
+                                                                                    .key(testKey)
+                                                                                    .build(),
+                                                                                AsyncResponseTransformer.toFile(testFile));
+
+        assertThat(response).succeedsWithin(Duration.of(10, ChronoUnit.SECONDS));
+        byte[] actualBody = Files.readAllBytes(testFile);
+        assertThat(actualBody).isEqualTo(expectedBody);
+        assertThat(response).isNotNull();
+        util.verifyCorrectAmountOfRequestsMade(numParts);
+    }
+
+    @Test
+    void errorOnFirstPart_nonRetryable() {
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey))).willReturn(
+            aResponse()
+                .withStatus(403)
+                .withBody("<Error><Code>AccessDenied</Code><Message>Test: Access denied!</Message></Error>")));
+
+        CompletableFuture<GetObjectResponse> resp = s3AsyncClient.getObject(b -> b
+                                                                                .bucket(testBucket)
+                                                                                .key(testKey)
+                                                                                .build(),
+                                                                            AsyncResponseTransformer.toFile(testFile));
+        assertThat(resp).failsWithin(Duration.of(10, ChronoUnit.SECONDS))
+                        .withThrowableOfType(ExecutionException.class)
+                        .withCauseInstanceOf(S3Exception.class)
+                        .withMessageContaining("Test: Access denied!");
+        verify(exactly(1), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey))));
+    }
+
+    @Test
+    void errorOnFirstPart_retryable() {
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("Started")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 1</Message></Error>"))
+                    .willSetStateTo("retry1"));
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry1")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 2</Message></Error>"))
+                    .willSetStateTo("retry2"));
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry2")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 3</Message></Error>"))
+                    .willSetStateTo("retry3"));
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry3")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 4</Message></Error>")));
+
+        CompletableFuture<GetObjectResponse> resp = s3AsyncClient.getObject(b -> b
+                                                                                .bucket(testBucket)
+                                                                                .key(testKey)
+                                                                                .build(),
+                                                                            AsyncResponseTransformer.toFile(testFile));
+        assertThat(resp).failsWithin(Duration.of(10, ChronoUnit.SECONDS))
+                        .withThrowableOfType(ExecutionException.class)
+                        .withCauseInstanceOf(S3Exception.class)
+                        .withMessageContaining("Internal error 4");
+        verify(exactly(4), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey))));
+    }
+
+    @Test
+    void errorOnMiddlePart_nonRetryable() {
+        util.stubForPart(testBucket, testKey, 1, 3, 1024);
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey))).willReturn(
+            aResponse()
+                .withStatus(403)
+                .withBody("<Error><Code>AccessDenied</Code><Message>Test: Access denied!</Message></Error>")));
+        util.stubForPart(testBucket, testKey, 3, 3, 1024);
+
+        CompletableFuture<GetObjectResponse> resp = s3AsyncClient.getObject(b -> b
+                                                                                .bucket(testBucket)
+                                                                                .key(testKey)
+                                                                                .build(),
+                                                                            AsyncResponseTransformer.toFile(testFile));
+
+        assertThat(resp).failsWithin(Duration.of(10, ChronoUnit.SECONDS))
+                        .withThrowableOfType(ExecutionException.class)
+                        .withCauseInstanceOf(S3Exception.class)
+                        .withMessageContaining("Test: Access denied!");
+        verify(exactly(1), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey))));
+        verify(exactly(1), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey))));
+    }
+
+    @Test
+    void errorOnMiddlePart_retryable() {
+        util.stubForPart(testBucket, testKey, 1, 3, 1024);
+        util.stubForPart(testBucket, testKey, 3, 3, 1024);
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("Started")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 1</Message></Error>"))
+                    .willSetStateTo("retry1"));
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry1")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 2</Message></Error>"))
+                    .willSetStateTo("retry2"));
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry2")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 3</Message></Error>"))
+                    .willSetStateTo("retry3")
+        );
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry3")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 4</Message></Error>")));
+
+        CompletableFuture<GetObjectResponse> resp = s3AsyncClient.getObject(b -> b
+                                                                                .bucket(testBucket)
+                                                                                .key(testKey)
+                                                                                .build(),
+                                                                            AsyncResponseTransformer.toFile(testFile));
+
+        assertThat(resp).failsWithin(Duration.of(10, ChronoUnit.SECONDS))
+                        .withThrowableOfType(ExecutionException.class)
+                        .withCauseInstanceOf(S3Exception.class)
+                        .withMessageContaining("Internal error 4");
+
+        verify(exactly(1), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey))));
+        verify(exactly(4), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey))));
+    }
+
+    @Test
+    void errorOnFirstPart_retryable_thenSucceeds() throws Exception {
+        int partSize = 1024;
+        int totalPart = 3;
+        Random random = new Random();
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("Started")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 1</Message></Error>"))
+                    .willSetStateTo("retry1"));
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry1")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 2</Message></Error>"))
+                    .willSetStateTo("retry2"));
+
+        byte[] part1Data = new byte[partSize];
+        random.nextBytes(part1Data);
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=%d", testBucket, testKey, 1)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry2")
+                    .willReturn(aResponse()
+                                    .withHeader("x-amz-mp-parts-count", totalPart + "")
+                                    .withHeader("x-amz-content-range", contentRangeHeader(1, totalPart, partSize))
+                                    .withHeader("ETag", "test-etag")
+                                    .withBody(part1Data)));
+
+        byte[] part2Data = new byte[partSize];
+        random.nextBytes(part2Data);
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=%d", testBucket, testKey, 2)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry2")
+                    .willReturn(aResponse()
+                                    .withHeader("x-amz-mp-parts-count", totalPart + "")
+                                    .withHeader("x-amz-content-range", contentRangeHeader(2, totalPart, partSize))
+                                    .withHeader("ETag", "test-etag")
+                                    .withBody(part2Data)));
+
+        byte[] part3Data = new byte[partSize];
+        random.nextBytes(part3Data);
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=%d", testBucket, testKey, 3)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry2")
+                    .willReturn(aResponse()
+                                    .withHeader("x-amz-mp-parts-count", totalPart + "")
+                                    .withHeader("x-amz-content-range", contentRangeHeader(3, totalPart, partSize))
+                                    .withHeader("ETag", "test-etag")
+                                    .withBody(part3Data)));
+
+
+    }
+
+    @Test
+    void errorOnMiddlePart_retryable_thenSucceeds() throws Exception {
+        int partSize = 1024;
+        int totalPart = 3;
+        Random random = new Random();
+        byte[] part1Data = util.stubForPart(testBucket, testKey, 1, totalPart, partSize);
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("Started")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 1</Message></Error>"))
+                    .willSetStateTo("retry1"));
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry1")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 2</Message></Error>"))
+                    .willSetStateTo("retry2"));
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry2")
+                    .willReturn(aResponse()
+                                    .withStatus(500)
+                                    .withBody("<Error><Code>InternalError</Code><Message>Internal error 3</Message></Error>"))
+                    .willSetStateTo("retry3")
+        );
+
+        byte[] part2Data = new byte[partSize];
+        random.nextBytes(part2Data);
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=%d", testBucket, testKey, 2)))
+                    .inScenario("retry")
+                    .whenScenarioStateIs("retry3")
+                    .willReturn(aResponse()
+                                    .withHeader("x-amz-mp-parts-count", totalPart + "")
+                                    .withHeader("x-amz-content-range", contentRangeHeader(2, totalPart, partSize))
+                                    .withHeader("ETag", "test-etag")
+                                    .withBody(part2Data)));
+
+        byte[] part3Data = new byte[partSize];
+        random.nextBytes(part3Data);
+
+        stubFor(get(urlEqualTo(String.format("/%s/%s?partNumber=%d", testBucket, testKey, 3)))
+                    .willReturn(aResponse()
+                                    .withHeader("x-amz-mp-parts-count", totalPart + "")
+                                    .withHeader("x-amz-content-range", contentRangeHeader(3, totalPart, partSize))
+                                    .withHeader("ETag", "test-etag")
+                                    .withBody(part3Data)));
+
+        CompletableFuture<GetObjectResponse> resp = s3AsyncClient.getObject(b -> b
+                                                                                .bucket(testBucket)
+                                                                                .key(testKey)
+                                                                                .build(),
+                                                                            AsyncResponseTransformer.toFile(testFile));
+
+        assertThat(resp).succeedsWithin(Duration.of(10, ChronoUnit.SECONDS));
+
+        verify(exactly(1), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=1", testBucket, testKey))));
+        verify(exactly(4), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=2", testBucket, testKey))));
+        verify(exactly(1), getRequestedFor(urlEqualTo(String.format("/%s/%s?partNumber=3", testBucket, testKey))));
+
+        assertThat(Files.exists(testFile)).isTrue();
+        byte[] actualBody = Files.readAllBytes(testFile);
+        byte[] expectedBody = new byte[partSize * totalPart];
+        System.arraycopy(part1Data, 0, expectedBody, 0, partSize);
+        System.arraycopy(part2Data, 0, expectedBody, partSize, partSize);
+        System.arraycopy(part3Data, 0, expectedBody, partSize * 2, partSize);
+        assertThat(actualBody).isEqualTo(expectedBody);
+
+    }
+
+    @Test
+    void veryHighPartCount_shouldSucceed() throws Exception {
+        int numParts = 5000;
+        int partSize = 100;
+
+        byte[] expectedBody = util.stubAllParts(testBucket, testKey, numParts, partSize);
+
+        CompletableFuture<GetObjectResponse> response = s3AsyncClient.getObject(b -> b
+                                                                                    .bucket(testBucket)
+                                                                                    .key(testKey)
+                                                                                    .build(),
+                                                                                AsyncResponseTransformer.toFile(testFile));
+
+        assertThat(response).succeedsWithin(Duration.of(5, ChronoUnit.MINUTES));
+        response.join();
+        byte[] actualBody = Files.readAllBytes(testFile);
+        assertThat(actualBody).isEqualTo(expectedBody);
+        assertThat(response).isNotNull();
+        util.verifyCorrectAmountOfRequestsMade(numParts);
+
+    }
+}
\ No newline at end of file
diff --git a/test/architecture-tests/archunit_store/4195d6e3-8849-4e5a-848d-04f810577cd3 b/test/architecture-tests/archunit_store/4195d6e3-8849-4e5a-848d-04f810577cd3
index 8c3b1f284548..c1c17fce7485 100644
--- a/test/architecture-tests/archunit_store/4195d6e3-8849-4e5a-848d-04f810577cd3
+++ b/test/architecture-tests/archunit_store/4195d6e3-8849-4e5a-848d-04f810577cd3
@@ -41,7 +41,7 @@ Method <software.amazon.awssdk.services.s3.internal.settingproviders.UseArnRegio
 Method <software.amazon.awssdk.services.sqs.internal.batchmanager.ReceiveSqsMessageHelper.clear()> calls method <software.amazon.awssdk.utils.Logger.warn(java.util.function.Supplier, java.lang.Throwable)> in (ReceiveSqsMessageHelper.java:132)
 Method <software.amazon.awssdk.transfer.s3.internal.AsyncBufferingSubscriber.onSubscribe(org.reactivestreams.Subscription)> calls method <software.amazon.awssdk.utils.Logger.warn(java.util.function.Supplier)> in (AsyncBufferingSubscriber.java:67)
 Method <software.amazon.awssdk.transfer.s3.internal.PauseResumeHelper.fileModified(software.amazon.awssdk.transfer.s3.model.ResumableFileUpload, software.amazon.awssdk.services.s3.S3AsyncClient)> calls method <software.amazon.awssdk.utils.Logger.warn(java.util.function.Supplier, java.lang.Throwable)> in (PauseResumeHelper.java:59)
-Method <software.amazon.awssdk.transfer.s3.internal.progress.ContentRangeParser.totalBytes(java.lang.String)> calls method <software.amazon.awssdk.utils.Logger.warn(java.util.function.Supplier, java.lang.Throwable)> in (ContentRangeParser.java:71)
+Method <software.amazon.awssdk.utils.ContentRangeParser.totalBytes(java.lang.String)> calls method <software.amazon.awssdk.utils.Logger.warn(java.util.function.Supplier, java.lang.Throwable)> in (ContentRangeParser.java:71)
 Method <software.amazon.awssdk.transfer.s3.progress.LoggingTransferListener.transferFailed(software.amazon.awssdk.transfer.s3.progress.TransferListener$Context$TransferFailed)> calls method <software.amazon.awssdk.utils.Logger.warn(java.util.function.Supplier, java.lang.Throwable)> in (LoggingTransferListener.java:76)
 Method <software.amazon.awssdk.utils.Logger.log(org.slf4j.event.Level, java.util.function.Supplier)> calls method <software.amazon.awssdk.utils.Logger.warn(java.util.function.Supplier)> in (Logger.java:205)
 Method <software.amazon.awssdk.utils.async.AddingTrailingDataSubscriber.onSubscribe(org.reactivestreams.Subscription)> calls method <software.amazon.awssdk.utils.Logger.warn(java.util.function.Supplier, java.lang.Throwable)> in (AddingTrailingDataSubscriber.java:73)
diff --git a/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/progress/ContentRangeParser.java b/utils/src/main/java/software/amazon/awssdk/utils/ContentRangeParser.java
similarity index 61%
rename from services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/progress/ContentRangeParser.java
rename to utils/src/main/java/software/amazon/awssdk/utils/ContentRangeParser.java
index 03e67c402a56..f21dd1702b25 100644
--- a/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/progress/ContentRangeParser.java
+++ b/utils/src/main/java/software/amazon/awssdk/utils/ContentRangeParser.java
@@ -13,12 +13,11 @@
  * permissions and limitations under the License.
  */
 
-package software.amazon.awssdk.transfer.s3.internal.progress;
+package software.amazon.awssdk.utils;
 
+import java.util.Optional;
 import java.util.OptionalLong;
 import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.utils.Logger;
-import software.amazon.awssdk.utils.StringUtils;
 
 /**
  * Parse a Content-Range header value into a total byte count. The expected format is the following: <p></p>
@@ -72,4 +71,46 @@ public static OptionalLong totalBytes(String contentRange) {
             return OptionalLong.empty();
         }
     }
+
+    /**
+     * Parse the Content-Range to extract the byte range from the content. Only supports the {@code bytes} unit, any
+     * other unit will result in an empty OptionalLong. If byte range in unknown, which is represented by a {@code *} symbol
+     * in the header value, an empty OptionalLong will be returned.
+     *
+     * @param contentRange the value of the Content-Range header to be parsed.
+     * @return The total number of bytes in the content range or an empty optional if the contentRange is null, empty or if the
+     * total length is not a valid long.
+     */
+    public static Optional<Pair<Long, Long>> range(String contentRange) {
+        if (StringUtils.isEmpty(contentRange)) {
+            return Optional.empty();
+        }
+
+        String trimmed = contentRange.trim();
+        if (!trimmed.startsWith("bytes ")) {
+            return Optional.empty();
+        }
+        String withoutBytes = trimmed.substring("bytes ".length());
+        if (withoutBytes.startsWith("*")) {
+            return Optional.empty();
+        }
+        int hyphen = withoutBytes.indexOf('-');
+        if (hyphen == -1) {
+            return Optional.empty();
+        }
+        String begin = withoutBytes.substring(0, hyphen);
+        int slash = withoutBytes.indexOf('/');
+        if (slash == -1) {
+            return Optional.empty();
+        }
+        String end = withoutBytes.substring(hyphen + 1, slash);
+        try {
+            long startInt = Long.parseLong(begin);
+            long endInt = Long.parseLong(end);
+            return Optional.of(Pair.of(startInt, endInt));
+        } catch (Exception e) {
+            return Optional.empty();
+        }
+    }
+
 }
diff --git a/utils/src/main/java/software/amazon/awssdk/utils/ToString.java b/utils/src/main/java/software/amazon/awssdk/utils/ToString.java
index 6b6fbbbe7a55..a066ae68e6e6 100644
--- a/utils/src/main/java/software/amazon/awssdk/utils/ToString.java
+++ b/utils/src/main/java/software/amazon/awssdk/utils/ToString.java
@@ -16,6 +16,7 @@
 package software.amazon.awssdk.utils;
 
 import java.util.Arrays;
+import java.util.concurrent.locks.ReentrantLock;
 import software.amazon.awssdk.annotations.NotThreadSafe;
 import software.amazon.awssdk.annotations.SdkProtectedApi;
 
diff --git a/ztest-hagrid.sh b/ztest-hagrid.sh
new file mode 100644
index 000000000000..8a97d9081648
--- /dev/null
+++ b/ztest-hagrid.sh
@@ -0,0 +1,12 @@
+#!/bin/bash
+
+dl='getHagridFile'
+ul='uploadHagridFile'
+
+key='48TB'
+path="/mnt/raid0/$key"
+
+mvn test -pl :s3-transfer-manager \
+    -Dtest=software.amazon.awssdk.transfer.s3.HagridTest#$ul \
+    -Dtestpath=$path \
+    -Dtestkey=$key
\ No newline at end of file