Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: update UploadCallable to use createFrom to avoid NPE trying to resolve resulting object #2086

Merged
merged 5 commits into from
Jun 23, 2023
Merged

fix: update UploadCallable to use createFrom to avoid NPE trying to resolve resulting object #2086

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3a04e5b
fix: Fix bug where the Storage Object is accessed before WriteChannel…
sydney-munro Jun 21, 2023
15b3d3f
Finish writing skipIfExists test
sydney-munro Jun 21, 2023
fb66e68
remove accidental commit
sydney-munro Jun 21, 2023
76462df
Additional tests and applySkipIfExists simplification
sydney-munro Jun 22, 2023
976b9ba
Use createFrom instead of managing the File/WriteChannels directly
sydney-munro Jun 22, 2023
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
11 changes: 10 additions & 1 deletion ...-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelUploadConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ private ParallelUploadConfig(
this.prefix = prefix;
this.bucketName = bucketName;
this.targetOptsPerRequest = targetOptsPerRequest;
this.writeOptsPerRequest = writeOptsPerRequest;
this.writeOptsPerRequest = applySkipIfExists(skipIfExists, writeOptsPerRequest);
}

/** If a corresponding object already exists skip uploading the object */
Expand Down Expand Up @@ -115,6 +115,15 @@ public static Builder newBuilder() {
return new Builder();
}

private static List<BlobWriteOption> applySkipIfExists(
boolean skipIfExists, List<BlobWriteOption> writeOptsPerRequest) {
if (skipIfExists) {
return ImmutableList.copyOf(
BlobWriteOption.dedupe(writeOptsPerRequest, BlobWriteOption.doesNotExist()));
}
return writeOptsPerRequest;
}

@BetaApi
public static final class Builder {

Expand Down
31 changes: 13 additions & 18 deletions ...-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/UploadCallable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,16 +16,12 @@

package com.google.cloud.storage.transfermanager;

import com.google.cloud.WriteChannel;
import com.google.cloud.storage.Blob;
import com.google.cloud.storage.BlobInfo;
import com.google.cloud.storage.PackagePrivateMethodWorkarounds;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.Storage.BlobWriteOption;
import com.google.common.io.ByteStreams;
import java.nio.channels.FileChannel;
import com.google.cloud.storage.StorageException;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.Optional;
import java.util.concurrent.Callable;

final class UploadCallable implements Callable<UploadResult> {
Expand Down Expand Up @@ -61,24 +57,23 @@ public UploadResult call() throws Exception {
}

private UploadResult uploadWithoutChunking() {
long bytesCopied = -1L;
try {
Optional<BlobInfo> newBlob;
try (FileChannel r = FileChannel.open(sourceFile, StandardOpenOption.READ);
WriteChannel w = storage.writer(originalBlob, opts)) {
w.setChunkSize(transferManagerConfig.getPerWorkerBufferSize());
bytesCopied = ByteStreams.copy(r, w);
newBlob = PackagePrivateMethodWorkarounds.maybeGetBlobInfoFunction().apply(w);
}
Blob from = storage.createFrom(originalBlob, sourceFile, opts);
return UploadResult.newBuilder(originalBlob, TransferStatus.SUCCESS)
.setUploadedBlob(newBlob.get())
.setUploadedBlob(from.asBlobInfo())
.build();
} catch (Exception e) {
if (bytesCopied == -1) {
return UploadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_START)
} catch (StorageException e) {
if (parallelUploadConfig.isSkipIfExists() && e.getCode() == 412) {
return UploadResult.newBuilder(originalBlob, TransferStatus.SKIPPED)
.setException(e)
.build();
} else {
// TODO: check for FAILED_TO_START conditions
return UploadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
.setException(e)
.build();
}
} catch (Exception e) {
return UploadResult.newBuilder(originalBlob, TransferStatus.FAILED_TO_FINISH)
.setException(e)
.build();
Expand Down
66 changes: 63 additions & 3 deletions google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,11 @@ public void uploadFiles() throws Exception {
UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
List<UploadResult> uploadResults = job.getUploadResults();
assertThat(uploadResults).hasSize(3);
assertThat(
uploadResults.stream()
.filter(result -> result.getStatus() == TransferStatus.SUCCESS)
.collect(Collectors.toList()))
.hasSize(3);
}
}

Expand All @@ -153,6 +158,11 @@ public void uploadFilesWithOpts() throws Exception {
UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
List<UploadResult> uploadResults = job.getUploadResults();
assertThat(uploadResults).hasSize(3);
assertThat(
uploadResults.stream()
.filter(result -> result.getStatus() == TransferStatus.SUCCESS)
.collect(Collectors.toList()))
.hasSize(3);
}
}

Expand All @@ -178,9 +188,14 @@ public void uploadFilesOneFailure() throws Exception {
assertThat(uploadResults).hasSize(4);
assertThat(
uploadResults.stream()
.filter(x -> x.getStatus() == TransferStatus.FAILED_TO_START)
.filter(x -> x.getStatus() == TransferStatus.FAILED_TO_FINISH)
.collect(Collectors.toList()))
.hasSize(1);
assertThat(
uploadResults.stream()
.filter(result -> result.getStatus() == TransferStatus.SUCCESS)
.collect(Collectors.toList()))
.hasSize(3);
}
}

Expand All @@ -196,7 +211,7 @@ public void uploadNonexistentBucket() throws Exception {
ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
List<UploadResult> uploadResults = job.getUploadResults();
assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.FAILED_TO_START);
assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.FAILED_TO_FINISH);
assertThat(uploadResults.get(0).getException()).isInstanceOf(StorageException.class);
}
}
Expand All @@ -212,11 +227,56 @@ public void uploadNonexistentFile() throws Exception {
ParallelUploadConfig.newBuilder().setBucketName(bucketName).build();
UploadJob job = transferManager.uploadFiles(files, parallelUploadConfig);
List<UploadResult> uploadResults = job.getUploadResults();
assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.FAILED_TO_START);
assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.FAILED_TO_FINISH);
assertThat(uploadResults.get(0).getException()).isInstanceOf(NoSuchFileException.class);
}
}

@Test
public void uploadFailsSkipIfExists() throws Exception {
TransferManagerConfig config =
TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
String bucketName = bucket.getName();
try (TransferManager transferManager = config.getService();
TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
ParallelUploadConfig parallelUploadConfig =
ParallelUploadConfig.newBuilder().setBucketName(bucketName).setSkipIfExists(true).build();
UploadJob jobInitUpload =
transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
List<UploadResult> uploadResults = jobInitUpload.getUploadResults();
assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.SUCCESS);
UploadJob failedSecondUpload =
transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
List<UploadResult> failedResult = failedSecondUpload.getUploadResults();
assertThat(failedResult.get(0).getStatus()).isEqualTo(TransferStatus.SKIPPED);
}
}

@Test
public void uploadSkipIfExistsGenerationOverride() throws Exception {
TransferManagerConfig config =
TransferManagerConfigTestingInstances.defaults(storage.getOptions()).toBuilder().build();
String bucketName = bucket.getName();
try (TransferManager transferManager = config.getService();
TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize)) {
ParallelUploadConfig parallelUploadConfig =
ParallelUploadConfig.newBuilder()
.setBucketName(bucketName)
.setSkipIfExists(true)
.setWriteOptsPerRequest(ImmutableList.of(BlobWriteOption.generationMatch(5L)))
.build();
assertThat(parallelUploadConfig.getWriteOptsPerRequest()).hasSize(1);
UploadJob jobInitUpload =
transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
List<UploadResult> uploadResults = jobInitUpload.getUploadResults();
assertThat(uploadResults.get(0).getStatus()).isEqualTo(TransferStatus.SUCCESS);
UploadJob failedSecondUpload =
transferManager.uploadFiles(ImmutableList.of(tmpFile.getPath()), parallelUploadConfig);
List<UploadResult> failedResult = failedSecondUpload.getUploadResults();
assertThat(failedResult.get(0).getStatus()).isEqualTo(TransferStatus.SKIPPED);
}
}

sydney-munro marked this conversation as resolved.
Show resolved Hide resolved
@Test
public void downloadBlobs() throws Exception {
TransferManagerConfig config =
Expand Down