Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .changes/next-release/bugfix-S3TransferManager-0ea8067.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
{
"type": "bugfix",
"category": "S3 Transfer Manager",
"contributor": "",
"description": "Fixed `S3TransferManager#resumeDownloadFile` so that a download paused with the Java-based multipart S3 async client and resumed from a serialized `ResumableFileDownload` restarts from the beginning if any parts had completed."
}
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@
import static software.amazon.awssdk.services.s3.internal.crt.DefaultS3CrtAsyncClient.RESPONSE_FILE_OPTION;
import static software.amazon.awssdk.services.s3.internal.crt.DefaultS3CrtAsyncClient.RESPONSE_FILE_PATH;
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.CRT_PAUSE_RESUME_TOKEN;
import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.canResumeDownload;
import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.hasUnresumableRange;
import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.isObjectOrFileModified;
import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.toCrtDownloadFileRequest;

import java.util.concurrent.CompletableFuture;
Expand Down Expand Up @@ -187,6 +187,12 @@ FileDownload doResumeDownloadFile(ResumableFileDownload resumableFileDownload) {
CompletableFuture<DownloadFileRequest> newDownloadFileRequestFuture = new CompletableFuture<>();
S3MetaRequestPauseObservable observable = new S3MetaRequestPauseObservable();
AtomicReference<DownloadFileRequest> requestToSaveOnPauseRef = new AtomicReference<>(originalDownloadRequest);
// Created before the HEAD is sent, so it records the destination file before this resume can write to it.
CrtFileDownload fileDownload = new CrtFileDownload(returnFuture,
new ResumeTransferProgress(progressFuture),
observable,
requestToSaveOnPauseRef::get,
resumableFileDownload);

CompletableFuture<HeadObjectResponse> headFuture =
s3AsyncClient.headObject(b -> b.bucket(getObjectRequest.bucket()).key(getObjectRequest.key()));
Expand All @@ -195,7 +201,7 @@ FileDownload doResumeDownloadFile(ResumableFileDownload resumableFileDownload) {
CompletableFutureUtils.forwardExceptionTo(returnFuture, headFuture);

headFuture.thenAccept(headObjectResponse -> {
boolean restartFromBeginning = !canResumeDownload(resumableFileDownload, headObjectResponse)
boolean restartFromBeginning = isObjectOrFileModified(resumableFileDownload, headObjectResponse)
|| hasCompletedParts(resumableFileDownload)
|| hasUnresumableRange(originalDownloadRequest,
resumableFileDownload.bytesTransferred(),
Expand Down Expand Up @@ -226,25 +232,23 @@ FileDownload doResumeDownloadFile(ResumableFileDownload resumableFileDownload) {
return null;
});

return new CrtFileDownload(returnFuture,
new ResumeTransferProgress(progressFuture),
observable,
requestToSaveOnPauseRef::get,
resumableFileDownload);
return fileDownload;
}

/**
* A download that was paused while fetching individual parts can leave gaps in the destination file, because parts are
* written at their own offsets rather than in order. The CRT-based client only ever appends to the end of the file, so
* such a download cannot be continued and has to start over. This only arises when a {@link ResumableFileDownload}
* produced by the Java-based transfer manager is resumed with the CRT-based one.
* such a download cannot be continued and has to start over. Completed parts come from the Java-based transfer manager,
* either directly or carried over by a CRT pause that has not yet written to the file. The Java-based transfer manager
* restarts in the same way when the token no longer has the state needed to continue the parts, for example after the
* token was serialized.
*/
private static boolean hasCompletedParts(ResumableFileDownload resumableFileDownload) {
if (resumableFileDownload.completedParts().isEmpty()) {
return false;
}
log.debug(() -> "The paused download had completed individual parts, which the CRT-based S3 client cannot continue "
+ "from. The SDK will download the S3 object from the beginning.");
log.debug(() -> "The paused download had completed individual parts, so the destination file may have gaps between "
+ "them. The SDK will download the S3 object from the beginning.");
return true;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -489,6 +489,11 @@ FileDownload doResumeDownloadFile(ResumableFileDownload resumableFileDownload) {
CompletableFuture<TransferProgress> progressFuture = new CompletableFuture<>();
CompletableFuture<DownloadFileRequest> newDownloadFileRequestFuture = new CompletableFuture<>();
AtomicReference<DownloadFileRequest> requestToSaveOnPauseRef = new AtomicReference<>(originalDownloadRequest);
// Created before the HEAD is sent, so it records the destination file before this resume can write to it.
DefaultFileDownload fileDownload = new DefaultFileDownload(returnFuture,
new ResumeTransferProgress(progressFuture),
requestToSaveOnPauseRef::get,
resumableFileDownload);

CompletableFuture<HeadObjectResponse> headFuture =
s3AsyncClient.headObject(b -> b.bucket(getObjectRequest.bucket()).key(getObjectRequest.key()));
Expand All @@ -515,10 +520,7 @@ FileDownload doResumeDownloadFile(ResumableFileDownload resumableFileDownload) {
return null;
});

return new DefaultFileDownload(returnFuture,
new ResumeTransferProgress(progressFuture),
requestToSaveOnPauseRef::get,
resumableFileDownload);
return fileDownload;
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@

package software.amazon.awssdk.transfer.s3.internal.model;

import static software.amazon.awssdk.transfer.s3.internal.utils.FileUtils.fileNotModified;

import java.io.File;
import java.time.Duration;
import java.time.Instant;
Expand Down Expand Up @@ -67,6 +69,8 @@ public final class CrtFileDownload implements FileDownload {
private final S3MetaRequestPauseObservable observable;
private final Supplier<DownloadFileRequest> requestSupplier;
private final ResumableFileDownload resumedDownload;
private final long destinationLengthAtResume;
private final Instant destinationLastModifiedAtResume;
private final Duration pauseTimeout;

public CrtFileDownload(CompletableFuture<CompletedFileDownload> completionFuture,
Expand All @@ -89,6 +93,9 @@ public CrtFileDownload(CompletableFuture<CompletedFileDownload> completionFuture
this.observable = Validate.paramNotNull(observable, "observable");
this.requestSupplier = Validate.paramNotNull(requestSupplier, "requestSupplier");
this.resumedDownload = resumedDownload;
File destination = resumedDownload == null ? null : resumedDownload.downloadFileRequest().destination().toFile();
this.destinationLengthAtResume = destination == null ? 0 : destination.length();
this.destinationLastModifiedAtResume = destination == null ? null : Instant.ofEpochMilli(destination.lastModified());
this.pauseTimeout = Validate.paramNotNull(pauseTimeout, "pauseTimeout");
this.resumableFileDownload = new Lazy<>(this::doPause);
}
Expand Down Expand Up @@ -144,6 +151,15 @@ private ResumeToken pauseMetaRequest() {
}

private ResumableFileDownload resumableFileDownload(ResumeToken token) {
// If the destination file is unchanged since this resume started, nothing has been written to it, so the resumed
// token is still accurate for it, including its completed parts. Returning it means the next resume decides from the
// same token this resume used.
if (resumedDownload != null
&& fileNotModified(destinationLengthAtResume, destinationLastModifiedAtResume,
resumedDownload.downloadFileRequest().destination())) {
return resumedDownload;
}

DownloadFileRequest request = requestSupplier.get();
File destination = request.destination().toFile();

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@

package software.amazon.awssdk.transfer.s3.internal.model;

import static software.amazon.awssdk.transfer.s3.internal.utils.FileUtils.fileNotModified;

import java.io.File;
import java.time.Instant;
import java.util.List;
Expand All @@ -40,6 +42,8 @@ public final class DefaultFileDownload implements FileDownload {
private final TransferProgress progress;
private final Supplier<DownloadFileRequest> requestSupplier;
private final ResumableFileDownload resumedDownload;
private final long destinationLengthAtResume;
private final Instant destinationLastModifiedAtResume;

public DefaultFileDownload(CompletableFuture<CompletedFileDownload> completedFileDownloadFuture,
TransferProgress progress,
Expand All @@ -50,6 +54,9 @@ public DefaultFileDownload(CompletableFuture<CompletedFileDownload> completedFil
this.requestSupplier = Validate.paramNotNull(requestSupplier, "requestSupplier");
this.resumableFileDownload = new Lazy<>(this::doPause);
this.resumedDownload = resumedDownload;
File destination = resumedDownload == null ? null : resumedDownload.downloadFileRequest().destination().toFile();
this.destinationLengthAtResume = destination == null ? 0 : destination.length();
this.destinationLastModifiedAtResume = destination == null ? null : Instant.ofEpochMilli(destination.lastModified());
}

@Override
Expand All @@ -65,6 +72,15 @@ public ResumableFileDownload pause() {
private ResumableFileDownload doPause() {
completionFuture.cancel(true);

// If the destination file is unchanged since this resume started, nothing has been written to it, so the resumed
// token is still accurate for it, including its completed parts. Returning it means the next resume decides from the
// same token this resume used.
if (resumedDownload != null
&& fileNotModified(destinationLengthAtResume, destinationLastModifiedAtResume,
resumedDownload.downloadFileRequest().destination())) {
return resumedDownload;
}

Instant s3objectLastModified = null;
String s3objectEtag = null;
Long totalSizeInBytes = null;
Expand Down
Loading
Loading