Skip to content

Commit 3405611

Browse files
BenWhiteheadBenWhitehead
authored
feat: allow specifying an expected object size for resumable operations. (#2661)
Update resumable upload failure detection to be more specific about classifying a request as SCENARIO_5 Fixes #2511
1 parent 380057b commit 3405611

17 files changed

+347
-32
lines changed

‎google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,7 @@ public WritableByteChannelSession<?, BlobInfo> writeSession(
114114
BidiWriteObjectRequest req = grpc.getBidiWriteObjectRequest(info, opts);
115115

116116
ApiFuture<BidiResumableWrite> startResumableWrite =
117-
grpc.startResumableWrite(grpcCallContext, req);
117+
grpc.startResumableWrite(grpcCallContext, req, opts);
118118
return ResumableMedia.gapic()
119119
.write()
120120
.bidiByteChannel(grpc.storageClient.bidiWriteObjectCallable())

‎google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -156,7 +156,7 @@ public WritableByteChannelSession<?, BlobInfo> writeSession(
156156
WriteObjectRequest req = grpc.getWriteObjectRequest(info, opts);
157157

158158
ApiFuture<ResumableWrite> startResumableWrite =
159-
grpc.startResumableWrite(grpcCallContext, req);
159+
grpc.startResumableWrite(grpcCallContext, req, opts);
160160
return ResumableMedia.gapic()
161161
.write()
162162
.byteChannel(

‎google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import com.google.api.gax.rpc.ApiException;
2525
import com.google.api.gax.rpc.ApiStreamObserver;
2626
import com.google.api.gax.rpc.BidiStreamingCallable;
27+
import com.google.api.gax.rpc.ErrorDetails;
2728
import com.google.api.gax.rpc.OutOfRangeException;
2829
import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
2930
import com.google.cloud.storage.Conversions.Decoder;
@@ -345,10 +346,17 @@ public void onNext(BidiWriteObjectResponse value) {
345346
public void onError(Throwable t) {
346347
if (t instanceof OutOfRangeException) {
347348
OutOfRangeException oore = (OutOfRangeException) t;
348-
clientDetectedError(
349-
ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
350-
ImmutableList.of(lastWrittenRequest), null, context, oore));
351-
} else if (t instanceof ApiException) {
349+
ErrorDetails ed = oore.getErrorDetails();
350+
if (!(ed != null
351+
&& ed.getErrorInfo() != null
352+
&& ed.getErrorInfo().getReason().equals("GRPC_MISMATCHED_UPLOAD_SIZE"))) {
353+
clientDetectedError(
354+
ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
355+
ImmutableList.of(lastWrittenRequest), null, context, oore));
356+
return;
357+
}
358+
}
359+
if (t instanceof ApiException) {
352360
// use StorageExceptions logic to translate from ApiException to our status codes ensuring
353361
// things fall in line with our retry handlers.
354362
// This is suboptimal, as it will initialize a second exception, however this is the

‎google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedChunkedResumableWritableByteChannel.java‎

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import com.google.api.gax.rpc.ApiException;
2525
import com.google.api.gax.rpc.ApiStreamObserver;
2626
import com.google.api.gax.rpc.ClientStreamingCallable;
27+
import com.google.api.gax.rpc.ErrorDetails;
2728
import com.google.api.gax.rpc.OutOfRangeException;
2829
import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
2930
import com.google.cloud.storage.Conversions.Decoder;
@@ -267,11 +268,18 @@ public void onError(Throwable t) {
267268
if (t instanceof OutOfRangeException) {
268269
OutOfRangeException oore = (OutOfRangeException) t;
269270
open = false;
270-
StorageException storageException =
271-
ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
272-
segments, null, context, oore);
273-
invocationHandle.setException(storageException);
274-
} else if (t instanceof ApiException) {
271+
ErrorDetails ed = oore.getErrorDetails();
272+
if (!(ed != null
273+
&& ed.getErrorInfo() != null
274+
&& ed.getErrorInfo().getReason().equals("GRPC_MISMATCHED_UPLOAD_SIZE"))) {
275+
StorageException storageException =
276+
ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
277+
segments, null, context, oore);
278+
invocationHandle.setException(storageException);
279+
return;
280+
}
281+
}
282+
if (t instanceof ApiException) {
275283
// use StorageExceptions logic to translate from ApiException to our status codes ensuring
276284
// things fall in line with our retry handlers.
277285
// This is suboptimal, as it will initialize a second exception, however this is the

‎google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121
import com.google.api.gax.rpc.BidiStreamingCallable;
2222
import com.google.api.gax.rpc.ClientStreamingCallable;
2323
import com.google.api.gax.rpc.UnaryCallable;
24+
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
25+
import com.google.cloud.storage.UnifiedOpts.Opts;
2426
import com.google.common.util.concurrent.MoreExecutors;
2527
import com.google.storage.v2.BidiWriteObjectRequest;
2628
import com.google.storage.v2.BidiWriteObjectResponse;
@@ -50,7 +52,8 @@ GapicBidiWritableByteChannelSessionBuilder bidiByteChannel(
5052

5153
ApiFuture<ResumableWrite> resumableWrite(
5254
UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> callable,
53-
WriteObjectRequest writeObjectRequest) {
55+
WriteObjectRequest writeObjectRequest,
56+
Opts<ObjectTargetOpt> opts) {
5457
StartResumableWriteRequest.Builder b = StartResumableWriteRequest.newBuilder();
5558
if (writeObjectRequest.hasWriteObjectSpec()) {
5659
b.setWriteObjectSpec(writeObjectRequest.getWriteObjectSpec());
@@ -61,7 +64,7 @@ ApiFuture<ResumableWrite> resumableWrite(
6164
if (writeObjectRequest.hasObjectChecksums()) {
6265
b.setObjectChecksums(writeObjectRequest.getObjectChecksums());
6366
}
64-
StartResumableWriteRequest req = b.build();
67+
StartResumableWriteRequest req = opts.startResumableWriteRequest().apply(b).build();
6568
Function<String, WriteObjectRequest> f =
6669
uploadId ->
6770
writeObjectRequest.toBuilder().clearWriteObjectSpec().setUploadId(uploadId).build();
@@ -80,7 +83,8 @@ ApiFuture<ResumableWrite> resumableWrite(
8083

8184
ApiFuture<BidiResumableWrite> bidiResumableWrite(
8285
UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> x,
83-
BidiWriteObjectRequest writeObjectRequest) {
86+
BidiWriteObjectRequest writeObjectRequest,
87+
Opts<ObjectTargetOpt> opts) {
8488
StartResumableWriteRequest.Builder b = StartResumableWriteRequest.newBuilder();
8589
if (writeObjectRequest.hasWriteObjectSpec()) {
8690
b.setWriteObjectSpec(writeObjectRequest.getWriteObjectSpec());
@@ -91,7 +95,7 @@ ApiFuture<BidiResumableWrite> bidiResumableWrite(
9195
if (writeObjectRequest.hasObjectChecksums()) {
9296
b.setObjectChecksums(writeObjectRequest.getObjectChecksums());
9397
}
94-
StartResumableWriteRequest req = b.build();
98+
StartResumableWriteRequest req = opts.startResumableWriteRequest().apply(b).build();
9599
Function<String, BidiWriteObjectRequest> f =
96100
uploadId ->
97101
writeObjectRequest.toBuilder().clearWriteObjectSpec().setUploadId(uploadId).build();

‎google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -320,7 +320,7 @@ public Blob internalCreateFrom(Path path, BlobInfo info, Opts<ObjectTargetOpt> o
320320
ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write =
321321
storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext);
322322

323-
ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req);
323+
ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req, opts);
324324
ApiFuture<GrpcResumableSession> session2 =
325325
ApiFutures.transform(
326326
start,
@@ -365,7 +365,7 @@ public Blob createFrom(
365365
opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
366366
WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
367367

368-
ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req);
368+
ApiFuture<ResumableWrite> start = startResumableWrite(grpcCallContext, req, opts);
369369

370370
BufferedWritableByteChannelSession<WriteObjectResponse> session =
371371
ResumableMedia.gapic()
@@ -790,7 +790,7 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options
790790
// in JSON, the starting of the resumable session happens before the invocation of write can
791791
// happen. Emulate the same thing here.
792792
// 1. create the future
793-
ApiFuture<ResumableWrite> startResumableWrite = startResumableWrite(grpcCallContext, req);
793+
ApiFuture<ResumableWrite> startResumableWrite = startResumableWrite(grpcCallContext, req, opts);
794794
// 2. await the result of the future
795795
ResumableWrite resumableWrite = ApiFutureUtils.await(startResumableWrite);
796796
// 3. wrap the result in another future container before constructing the BlobWriteChannel
@@ -1919,7 +1919,7 @@ private UnbufferedReadableByteChannelSession<Object> unbufferedReadSession(
19191919

19201920
@VisibleForTesting
19211921
ApiFuture<ResumableWrite> startResumableWrite(
1922-
GrpcCallContext grpcCallContext, WriteObjectRequest req) {
1922+
GrpcCallContext grpcCallContext, WriteObjectRequest req, Opts<ObjectTargetOpt> opts) {
19231923
Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(req));
19241924
GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
19251925
return ResumableMedia.gapic()
@@ -1928,11 +1928,12 @@ ApiFuture<ResumableWrite> startResumableWrite(
19281928
storageClient
19291929
.startResumableWriteCallable()
19301930
.withDefaultCallContext(merge.withRetryableCodes(codes)),
1931-
req);
1931+
req,
1932+
opts);
19321933
}
19331934

19341935
ApiFuture<BidiResumableWrite> startResumableWrite(
1935-
GrpcCallContext grpcCallContext, BidiWriteObjectRequest req) {
1936+
GrpcCallContext grpcCallContext, BidiWriteObjectRequest req, Opts<ObjectTargetOpt> opts) {
19361937
Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(req));
19371938
GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
19381939
return ResumableMedia.gapic()
@@ -1941,7 +1942,8 @@ ApiFuture<BidiResumableWrite> startResumableWrite(
19411942
storageClient
19421943
.startResumableWriteCallable()
19431944
.withDefaultCallContext(merge.withRetryableCodes(codes)),
1944-
req);
1945+
req,
1946+
opts);
19451947
}
19461948

19471949
private SourceObject sourceObjectEncode(SourceBlob from) {

‎google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -190,7 +190,7 @@ public WritableByteChannelSession<?, BlobInfo> writeSession(
190190
opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
191191
ApiFuture<ResumableWrite> f =
192192
grpcStorage.startResumableWrite(
193-
grpcCallContext, grpcStorage.getWriteObjectRequest(info, opts));
193+
grpcCallContext, grpcStorage.getWriteObjectRequest(info, opts), opts);
194194
ApiFuture<WriteCtx<ResumableWrite>> start =
195195
ApiFutures.transform(f, WriteCtx::new, MoreExecutors.directExecutor());
196196

‎google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -205,7 +205,8 @@ public void rewindTo(long offset) {
205205
&& contentLength != null
206206
&& contentLength > 0) {
207207
String errorMessage = cause.getContent().toLowerCase(Locale.US);
208-
if (errorMessage.contains("content-range")) {
208+
if (errorMessage.contains("content-range")
209+
&& !errorMessage.contains("earlier")) { // TODO: exclude "earlier request"
209210
StorageException se =
210211
ResumableSessionFailureScenario.SCENARIO_5.toStorageException(
211212
uploadId, response, cause, cause::getContent);

‎google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
4444
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
4545
import com.google.cloud.storage.UnifiedOpts.Opts;
46+
import com.google.cloud.storage.UnifiedOpts.ResumableUploadExpectedObjectSize;
4647
import com.google.cloud.storage.UnifiedOpts.SourceGenerationMatch;
4748
import com.google.cloud.storage.UnifiedOpts.SourceGenerationNotMatch;
4849
import com.google.cloud.storage.UnifiedOpts.SourceMetagenerationMatch;
@@ -102,7 +103,8 @@ final class ParallelCompositeUploadWritableByteChannel implements BufferedWritab
102103
|| o instanceof SourceMetagenerationMatch
103104
|| o instanceof SourceMetagenerationNotMatch
104105
|| o instanceof Crc32cMatch
105-
|| o instanceof Md5Match;
106+
|| o instanceof Md5Match
107+
|| o instanceof ResumableUploadExpectedObjectSize;
106108
TO_EXCLUDE_FROM_PARTS = tmp.negate();
107109
}
108110

‎google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1352,6 +1352,23 @@ public static BlobWriteOption detectContentType() {
13521352
return new BlobWriteOption(UnifiedOpts.detectContentType());
13531353
}
13541354

1355+
/**
1356+
* Set a precondition on the number of bytes that GCS should expect for a resumable upload. See
1357+
* the docs for <a
1358+
* href="https://cloud.google.com/storage/docs/json_api/v1/parameters#xuploadcontentlength">X-Upload-Content-Length</a>
1359+
* for more detail.
1360+
*
1361+
* <p>If the method invoked with this option does not perform a resumable upload, this option
1362+
* will be ignored.
1363+
*
1364+
* @since 2.42.0
1365+
*/
1366+
@BetaApi
1367+
@TransportCompatibility({Transport.HTTP, Transport.GRPC})
1368+
public static BlobWriteOption expectedObjectSize(long objectContentSize) {
1369+
return new BlobWriteOption(UnifiedOpts.resumableUploadExpectedObjectSize(objectContentSize));
1370+
}
1371+
13551372
/**
13561373
* Deduplicate any options which are the same parameter. The value which comes last in {@code
13571374
* os} will be the value included in the return.

0 commit comments

Comments
 (0)