Skip to content

feat(gax): retry chunk upload on transient errors #14422

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

Merged
whowes merged 1 commit into from
Sep 25, 2026
Merged

feat(gax): retry chunk upload on transient errors #14422

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
40 changes: 39 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 @@ -35,9 +35,16 @@
import com.google.api.core.ApiFutures;
import com.google.api.core.BetaApi;
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.ResumableUploadClient;
import com.google.api.gax.resumable.ResumableUploadSession;
import com.google.api.gax.retrying.ExponentialRetryAlgorithm;
import com.google.api.gax.retrying.RetryAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.ScheduledRetryingExecutor;
import java.io.InputStream;
import java.time.Duration;
import org.jspecify.annotations.NullMarked;
import org.jspecify.annotations.Nullable;

Expand All @@ -54,9 +61,23 @@
public class ResumableUploadCallableImpl<RequestT, ResponseT>
extends ResumableUploadCallable<RequestT, ResponseT> {

private static final RetrySettings RETRY_SETTINGS =
RetrySettings.newBuilder()

@blakeli0 blakeli0 Sep 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should at least set initialRpcTimeoutDuration and totalTimeoutDuration. Otherwise the call could hang indefinitely (before global timeout kicks in). These are the default generated retrySetting values.

Separately, check if there is a cross-language sensible default.

@whowes whowes Sep 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The cross-language docs aren't very prescriptive here - no suggested values that I could see.

I did take a look at the Python resumable upload implementation and set initialRpcTimeoutDuration to 1m and totalTimeoutDuration to 2m as a result. I also adjusted the backoff params to mirror Python's too (now 1s initial delay, 2x multiplier, 1m max delay) - the former settings I had looked like they'd exhaust the retries in < 1s if the wire commands were failing fast.

.setInitialRetryDelayDuration(Duration.ofSeconds(1))
.setRetryDelayMultiplier(2.0)
.setMaxRetryDelayDuration(Duration.ofSeconds(60))
.setMaxAttempts(5)
.setInitialRpcTimeoutDuration(Duration.ofMinutes(1))
.setRpcTimeoutMultiplier(1.0)
.setMaxRpcTimeoutDuration(Duration.ofMinutes(1))
.setTotalTimeoutDuration(Duration.ofMinutes(2))
.build();

private final ResumableUploadClient<RequestT, ResponseT> client;
private final ResumableUploadCallSettings defaultCallSettings;
private final ClientContext clientContext;
private final UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>>
retryingUploadChunkCallable;

public ResumableUploadCallableImpl(
ResumableUploadClient<RequestT, ResponseT> client,
Expand All @@ -66,6 +87,9 @@
this.defaultCallSettings =
checkNotNull(defaultCallSettings, "defaultCallSettings must not be null");
this.clientContext = checkNotNull(clientContext, "clientContext must not be null");
this.retryingUploadChunkCallable =
createRetryingCallable(
client.uploadChunkCallable(), ResumableUploadCommand.UPLOAD, clientContext);
}

@Override
Expand All @@ -88,7 +112,7 @@

return ResumableUploadFutureImpl.create(
startFuture,
client.uploadChunkCallable(),
retryingUploadChunkCallable,
payload,
effectiveSettings,
clientContext.getDefaultCallContext());
Expand All @@ -99,4 +123,18 @@
String sessionUrl, InputStream payload, @Nullable ResumableUploadCallSettings settings) {
throw new UnsupportedOperationException("Session resumption is not yet implemented.");
}

private static <RequestT, ResponseT> UnaryCallable<RequestT, ResponseT> createRetryingCallable(

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

View check run for this annotation

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

Rename this generic name to match the regular expression '^[A-Z][0-9]?$'. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaDJ2VzQMdQCfkCO3QkW&open=AaDJ2VzQMdQCfkCO3QkW&pullRequest=14422

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

View check run for this annotation

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

Rename this generic name to match the regular expression '^[A-Z][0-9]?$'. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaDJ2VzPMdQCfkCO3QkV&open=AaDJ2VzPMdQCfkCO3QkV&pullRequest=14422
UnaryCallable<RequestT, ResponseT> callable,
ResumableUploadCommand command,
ClientContext clientContext) {
RetryAlgorithm<ResponseT> retryAlgorithm =
new RetryAlgorithm<>(
new ResumableUploadResultRetryAlgorithm<>(command),
new ExponentialRetryAlgorithm(RETRY_SETTINGS, clientContext.getClock()));
return new RetryingCallable<>(
clientContext.getDefaultCallContext(),
checkNotNull(callable, "callable must not be null"),
new ScheduledRetryingExecutor<>(retryAlgorithm, clientContext.getExecutor()));
}
}
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 @@ -214,6 +214,8 @@
void testResumableUploadCallable() {
ResumableUploadClient<String, String> uploadClient =
mock(ResumableUploadClient.class, Mockito.withSettings().withoutAnnotations());
when(uploadClient.uploadChunkCallable())
.thenReturn(mock(UnaryCallable.class, Mockito.withSettings().withoutAnnotations()));

Check warning on line 218 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=AaDGFYWiFZhPOSYKhVZo&open=AaDGFYWiFZhPOSYKhVZo&pullRequest=14422
ResumableUploadCallSettings settings =
ResumableUploadCallSettings.newBuilder().setChunkSize(1024).build();

Expand Down
106 changes: 106 additions & 0 deletions ...va/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -457,6 +457,102 @@ void testResumeCall_throwsUnsupportedOperationException() {
() -> callable.resumeCall("https://upload.url/session", streamOf("data"), null));
}

@Test
void testChunkRetry_transientFailureThenSuccess_retriesAndSucceeds() throws Exception {
stubStartSession("https://upload.url/chunk-retry-ok");
TrackableStream stream = new TrackableStream("01234567"); // exactly 1 chunk of 8 bytes
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
.thenReturn(
ApiFutures.immediateFailedFuture(createApiException(503, StatusCode.Code.UNAVAILABLE)))
.thenReturn(
ApiFutures.immediateFuture(
ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "chunk-done")));

ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream, null);

assertThat(future.get()).isEqualTo("chunk-done");
assertThat(future.isDone()).isTrue();
assertThat(stream.totalBytesRead).isEqualTo(8);
assertThat(stream.closed).isTrue();

ArgumentCaptor<ChunkUploadRequest> captor = ArgumentCaptor.forClass(ChunkUploadRequest.class);
verify(mockChunkCallable, times(2)).futureCall(captor.capture(), any());
List<ChunkUploadRequest> requests = captor.getAllValues();
assertThat(requests.get(0).getOffset()).isEqualTo(0);
assertThat(requests.get(0).getPayload()).isEqualTo("01234567".getBytes(StandardCharsets.UTF_8));
assertThat(requests.get(1).getOffset()).isEqualTo(0);
assertThat(requests.get(1).getPayload()).isEqualTo("01234567".getBytes(StandardCharsets.UTF_8));
}

@Test
void testChunkRetry_transientFailureExhaustion_surfacesLastError() {
stubStartSession("https://upload.url/chunk-exhaustion");
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
.thenReturn(
ApiFutures.immediateFailedFuture(createApiException(503, StatusCode.Code.UNAVAILABLE)));

ResumableUploadFuture<String> future =
callable.futureCall("resource-path", streamOf("hello"), null);

ExecutionException exception = assertThrows(ExecutionException.class, future::get);
assertThat(exception.getCause()).isInstanceOf(ApiException.class);
assertThat(((ApiException) exception.getCause()).getStatusCode().getTransportCode())
.isEqualTo(503);

// Default chunk retry settings has maxAttempts = 5
verify(mockChunkCallable, times(5)).futureCall(any(), any());
}

@Test
void testChunkRetry_cancellationDuringBackoff_deschedulesPendingAttempt() {
stubStartSession("https://upload.url/cancel-backoff");
SettableApiFuture<ChunkUploadResponse<String>> chunkAttempt0Future = SettableApiFuture.create();
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
.thenReturn(chunkAttempt0Future)
.thenReturn(
ApiFutures.immediateFuture(
ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "should-not-reach")));

ResumableUploadFuture<String> sessionFuture =
callable.futureCall("resource-path", streamOf("hello"), null);

// Fail attempt 0 with 503 to schedule backoff
chunkAttempt0Future.setException(createApiException(503, StatusCode.Code.UNAVAILABLE));

// Cancel while backoff is pending
assertThat(sessionFuture.cancel(true)).isTrue();
assertThat(sessionFuture.isCancelled()).isTrue();
assertThrows(CancellationException.class, sessionFuture::get);

// Only attempt 0 occurred; attempt 1 was de-scheduled
verify(mockChunkCallable, times(1)).futureCall(any(), any());
}

private static class HttpStatusStatusCode implements StatusCode {
private final int httpStatus;
private final StatusCode.Code code;

HttpStatusStatusCode(int httpStatus, StatusCode.Code code) {
this.httpStatus = httpStatus;
this.code = code;
}

@Override
public StatusCode.Code getCode() {
return code;
}

@Override
public Integer getTransportCode() {
return httpStatus;
}
}

private static ApiException createApiException(int httpStatus, StatusCode.Code code) {
return ApiExceptionFactory.createException(
"HTTP " + httpStatus, null, new HttpStatusStatusCode(httpStatus, code), false);
}

private void stubStartSession(String uploadUrl) {
when(mockStartCallable.futureCall(any(), any()))
.thenReturn(
Expand All @@ -477,11 +573,21 @@ private static void assertChunk(

private static class TrackableStream extends ByteArrayInputStream {
boolean closed = false;
int totalBytesRead = 0;

TrackableStream(String content) {
super(content.getBytes(StandardCharsets.UTF_8));
}

@Override
public int read(byte[] b, int off, int len) {
int read = super.read(b, off, len);
if (read > 0) {
totalBytesRead += read;
}
return read;
}

@Override
public void close() throws IOException {
closed = true;
Expand Down
Loading