Skip to content

feat(gax): add chunk upload recovery loop for resumable uploads #14424

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

Draft
whowes wants to merge 1 commit into
base: whowes/resumable-upload-buffer-window
Choose a base branch
from
Draft

feat(gax): add chunk upload recovery loop for resumable uploads #14424

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
13 changes: 12 additions & 1 deletion ...m-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@
import com.google.api.core.InternalApi;
import com.google.api.gax.resumable.ChunkUploadRequest;
import com.google.api.gax.resumable.ChunkUploadResponse;
import com.google.api.gax.resumable.QueryStatusRequest;
import com.google.api.gax.resumable.QueryStatusResponse;
import com.google.api.gax.resumable.ResumableUploadClient;
import com.google.api.gax.resumable.ResumableUploadSession;
import com.google.api.gax.retrying.ExponentialRetryAlgorithm;
Expand Down Expand Up @@ -74,6 +76,8 @@ public class ResumableUploadCallableImpl<RequestT, ResponseT>
private final ClientContext clientContext;
private final UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>>
retryingUploadChunkCallable;
private final UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>>
retryingQueryCallable;

public ResumableUploadCallableImpl(
ResumableUploadClient<RequestT, ResponseT> client,
Expand All @@ -89,6 +93,8 @@ public ResumableUploadCallableImpl(
.build();
this.retryingUploadChunkCallable =
createRetryingCallable(client.uploadChunkCallable(), ResumableUploadCommand.UPLOAD);
this.retryingQueryCallable =
createRetryingCallable(client.queryStatusCallable(), ResumableUploadCommand.QUERY);
}

@Override
Expand All @@ -110,7 +116,12 @@ public ResumableUploadFuture<ResponseT> futureCall(
}

return ResumableUploadFutureImpl.create(
startFuture, retryingUploadChunkCallable, payload, effectiveSettings, clientContext);
startFuture,
retryingUploadChunkCallable,
retryingQueryCallable,
payload,
effectiveSettings,
clientContext);
}

@Override
Expand Down
55 changes: 26 additions & 29 deletions ...va/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,24 +210,22 @@
if (result.isDone()) {
return;
}
try {
if (response.getUploadStatus() == ResumableUploadStatus.FINAL) {
result.set(response.getResponse());
} else if (buffer.isFinal()) {
throw new IllegalStateException(
"Upload stream ended and final chunk was transmitted, but server returned incomplete"
+ " status for upload URL: "
+ uploadUrl);
} else {
long nextOffset = buffer.getBufferBaseOffset() + buffer.getPayloadLength();
transmitChunk(nextOffset);
}
} catch (Throwable t) {
result.setException(t);
long nextOffset = buffer.getBufferBaseOffset() + buffer.getPayloadLength();
if (response.getUploadStatus() == ResumableUploadStatus.FINAL) {
result.set(response.getResponse());
} else if (buffer.isFinal()) {
result.setException(
new IllegalStateException(
"Upload stream ended and final chunk was transmitted, but server returned"
+ " incomplete status for upload URL: "
+ uploadUrl));
} else {
chunkExecutor.execute(() -> transmitChunk(nextOffset));
}
}

private ChunkUploadRequest buildCurrentChunkRequest() {
// Determine if this is the final chunk and build the chunk request.
return ChunkUploadRequest.newBuilder()
.setUploadUrl(uploadUrl)
.setPayload(buffer.getBuffer())
Expand All @@ -237,21 +235,20 @@
.build();
}

private static FailedPreconditionException protocolViolation(String message) {
return new FailedPreconditionException(
message,
null,
new StatusCode() {
@Override
public StatusCode.Code getCode() {
return StatusCode.Code.FAILED_PRECONDITION;
}
private static final StatusCode FAILED_PRECONDITION_STATUS_CODE =
new StatusCode() {
@Override
public StatusCode.Code getCode() {
return StatusCode.Code.FAILED_PRECONDITION;
}

@Override
public @Nullable Object getTransportCode() {
return null;
}
},
false);
@Override
public @Nullable Object getTransportCode() {
return null;
}
};

private static FailedPreconditionException protocolViolation(String message) {
return new FailedPreconditionException(message, null, FAILED_PRECONDITION_STATUS_CODE, false);

Check warning on line 252 in sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java

View check run for this annotation

SonarQubeCloud / [gapic-generator-java-root] SonarCloud Code Analysis

Annotate the parameter with @javax.annotation.Nullable in constructor declaration, or make sure that null can not be passed as argument. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaDIKOYdVmg9UZRlUUaz&open=AaDIKOYdVmg9UZRlUUaz&pullRequest=14424
}
}
16 changes: 15 additions & 1 deletion ...orm-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@
import com.google.api.core.SettableApiFuture;
import com.google.api.gax.resumable.ChunkUploadRequest;
import com.google.api.gax.resumable.ChunkUploadResponse;
import com.google.api.gax.resumable.QueryStatusRequest;
import com.google.api.gax.resumable.QueryStatusResponse;
import com.google.api.gax.resumable.ResumableUploadSession;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.errorprone.annotations.concurrent.GuardedBy;
Expand Down Expand Up @@ -65,6 +67,8 @@ final class ResumableUploadFutureImpl<ResponseT> implements ResumableUploadFutur
private final ApiFuture<ResumableUploadSession> startFuture;
private final UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>>
uploadChunkCallable;
private final UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>>
queryStatusCallable;
private final InputStream payload;
private final ResumableUploadCallSettings settings;
private final ClientContext clientContext;
Expand All @@ -85,12 +89,18 @@ final class ResumableUploadFutureImpl<ResponseT> implements ResumableUploadFutur
static <ResponseT> ResumableUploadFutureImpl<ResponseT> create(
ApiFuture<ResumableUploadSession> startFuture,
UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> uploadChunkCallable,
UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
InputStream payload,
ResumableUploadCallSettings settings,
ClientContext clientContext) {
ResumableUploadFutureImpl<ResponseT> future =
new ResumableUploadFutureImpl<>(
startFuture, uploadChunkCallable, payload, settings, clientContext);
startFuture,
uploadChunkCallable,
queryStatusCallable,
payload,
settings,
clientContext);
try {
future.start();
} catch (Throwable t) {
Expand All @@ -102,12 +112,15 @@ static <ResponseT> ResumableUploadFutureImpl<ResponseT> create(
private ResumableUploadFutureImpl(
ApiFuture<ResumableUploadSession> startFuture,
UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> uploadChunkCallable,
UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
InputStream payload,
ResumableUploadCallSettings settings,
ClientContext clientContext) {
this.startFuture = checkNotNull(startFuture, "startFuture must not be null");
this.uploadChunkCallable =
checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
this.queryStatusCallable =
checkNotNull(queryStatusCallable, "queryStatusCallable must not be null");
this.payload = checkNotNull(payload, "payload must not be null");
this.settings = checkNotNull(settings, "settings must not be null");
checkArgument(settings.getChunkSize() > 0, "chunkSize must be > 0");
Expand All @@ -128,6 +141,7 @@ public void onSuccess(ResumableUploadSession session) {
ResumableUploadChunkCoordinator<ResponseT> coordinator =
new ResumableUploadChunkCoordinator<>(
uploadChunkCallable,
queryStatusCallable,
uploadSessionUrl,
payload,
settings.getChunkSize(),
Expand Down
2 changes: 2 additions & 0 deletions sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/CallableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,8 @@
mock(ResumableUploadClient.class, Mockito.withSettings().withoutAnnotations());
when(uploadClient.uploadChunkCallable())
.thenReturn(mock(UnaryCallable.class, Mockito.withSettings().withoutAnnotations()));
when(uploadClient.queryStatusCallable())
.thenReturn(mock(UnaryCallable.class, Mockito.withSettings().withoutAnnotations()));

Check warning on line 220 in sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/CallableTest.java

View check run for this annotation

SonarQubeCloud / [gapic-generator-java-root] SonarCloud Code Analysis

Extract this mock creation to a local variable. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaCyXfHtzZY4JqY3jcOe&open=AaCyXfHtzZY4JqY3jcOe&pullRequest=14424
ResumableUploadCallSettings settings =
ResumableUploadCallSettings.newBuilder().setChunkSize(1024).build();

Expand Down
Loading
Loading