Skip to content

feat(gax): add ApiCallContext and request-level settings overloads to ResumableUploadCallable #14251

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 from
Draft

feat(gax): add ApiCallContext and request-level settings overloads to ResumableUploadCallable #14251

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
62 changes: 59 additions & 3 deletions ...tform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,19 +48,75 @@ public abstract class ResumableUploadCallable<RequestT, ResponseT> {
protected ResumableUploadCallable() {}

/**
* Performs a new resumable upload asynchronously.
* Performs a new resumable upload asynchronously with default call context and default settings.
*
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
* cancellation.
*
* @param request the request message
* @param payload the data payload input stream to upload and close
* @param settings call settings overrides; may be {@code null}
* @return future for tracking and controlling the upload
*/
public ResumableUploadFuture<ResponseT> futureCall(RequestT request, InputStream payload) {
return futureCall(request, payload, null, null);
}

/**
* Performs a new resumable upload asynchronously with a call context override and default
* settings.
*
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
* cancellation.
*
* @param request the request message
* @param payload the data payload input stream to upload and close
* @param context call context overrides (e.g. extra headers, credentials, timeout); may be {@code
* null}
* @return future for tracking and controlling the upload
*/
public ResumableUploadFuture<ResponseT> futureCall(
RequestT request, InputStream payload, @Nullable ApiCallContext context) {
return futureCall(request, payload, context, null);
}

/**
* Performs a new resumable upload asynchronously with settings overrides and default call
* context.
*
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
* cancellation.
*
* @param request the request message
* @param payload the data payload input stream to upload and close
* @param settings request-level call settings overrides; may be {@code null}
* @return future for tracking and controlling the upload
*/
public ResumableUploadFuture<ResponseT> futureCall(
RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings) {
return futureCall(request, payload, null, settings);
}

/**
* Performs a new resumable upload asynchronously with call context and settings overrides.
*
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
* cancellation.
*
* @param request the request message
* @param payload the data payload input stream to upload and close
* @param context call context overrides; may be {@code null}
* @param settings request-level call settings overrides; may be {@code null}
* @return future for tracking and controlling the upload
*/
public abstract ResumableUploadFuture<ResponseT> futureCall(
RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings);
RequestT request,
InputStream payload,
@Nullable ApiCallContext context,
@Nullable ResumableUploadCallSettings settings);
Comment on lines 115 to +119

@gemini-code-assist gemini-code-assist Bot Sep 3, 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.

high

Removing the 3-parameter futureCall overload that accepts ResumableUploadCallSettings breaks backward compatibility for existing callers and subclasses. To maintain source and binary compatibility, we should keep the 3-parameter overload as a concrete method that delegates to the new 4-parameter method with a null context.

  /**
   * Performs a new resumable upload asynchronously with default call context and settings overrides.
   *
   * @param request the request message
   * @param payload the data payload input stream to upload and close
   * @param settings request-level call settings overrides; may be {@code null}
   * @return future for tracking and controlling the upload
   */
  public ResumableUploadFuture<ResponseT> futureCall(
      RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings) {
    return futureCall(request, payload, null, settings);
  }

  /**
   * Performs a new resumable upload asynchronously with call context and settings overrides.
   *
   * @param request the request message
   * @param payload the data payload input stream to upload and close
   * @param context call context overrides; may be {@code null}
   * @param settings request-level call settings overrides; may be {@code null}
   * @return future for tracking and controlling the upload
   */
  public abstract ResumableUploadFuture<ResponseT> futureCall(
      RequestT request,
      InputStream payload,
      @Nullable ApiCallContext context,
      @Nullable ResumableUploadCallSettings settings);


/**
* Resumes an existing resumable upload session asynchronously using a saved session URL.
Expand Down
8 changes: 6 additions & 2 deletions ...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 @@ -75,14 +75,18 @@

@Override
public ResumableUploadFuture<ResponseT> futureCall(
RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings) {
RequestT request,
InputStream payload,
@Nullable ApiCallContext context,
@Nullable ResumableUploadCallSettings settings) {
checkNotNull(request, "request must not be null");
checkNotNull(payload, "payload must not be null");
ResumableUploadCallSettings effectiveSettings = defaultCallSettings.merge(settings);
ApiCallContext effectiveCallContext = defaultCallContext.merge(context);

Check warning on line 85 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

Annotate the parameter with @javax.annotation.Nullable in method 'merge' 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=AaBlzKp1mCL_HGZD9DxA&open=AaBlzKp1mCL_HGZD9DxA&pullRequest=14251

ApiFuture<ResumableUploadSession> startFuture;
try {
startFuture = client.startUploadCallable().futureCall(request, defaultCallContext);
startFuture = client.startUploadCallable().futureCall(request, effectiveCallContext);
} catch (Throwable t) {
startFuture = ApiFutures.immediateFailedFuture(t);
}
Expand Down
120 changes: 106 additions & 14 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 @@ -112,7 +112,7 @@
ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-single")));

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

assertThat(future.get()).isEqualTo("response-single");
assertThat(future.isDone()).isTrue();
Expand All @@ -136,7 +136,7 @@
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-multi")));

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

assertThat(future.get()).isEqualTo("response-multi");
assertThat(future.isDone()).isTrue();
Expand All @@ -156,7 +156,7 @@
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-zero")));

ResumableUploadFuture<String> future =
callable.futureCall("resource-path", new ByteArrayInputStream(new byte[0]), null);
callable.futureCall("resource-path", new ByteArrayInputStream(new byte[0]));

assertThat(future.get()).isEqualTo("response-zero");

Expand All @@ -176,7 +176,7 @@
ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-exact-single")));

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

assertThat(future.get()).isEqualTo("response-exact-single");

Expand All @@ -194,7 +194,7 @@
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, null)));

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

assertThat(future.get()).isNull();
assertThat(future.isDone()).isTrue();
Expand All @@ -217,7 +217,7 @@
});

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

assertThat(chunkStarted.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(future.cancel(true)).isTrue();
Expand All @@ -233,7 +233,7 @@
SettableApiFuture<ResumableUploadSession> startFuture = SettableApiFuture.create();
when(mockStartCallable.futureCall(any(), any())).thenReturn(startFuture);
ResumableUploadFuture<String> future =
callable.futureCall("resource-path", streamOf("data"), null);
callable.futureCall("resource-path", streamOf("data"));
assertThat(future.cancel(true)).isTrue();
assertThat(future.isCancelled()).isTrue();

Expand All @@ -248,7 +248,7 @@
.thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("start failed")));

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

ExecutionException exception = assertThrows(ExecutionException.class, future::get);
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
Expand All @@ -263,7 +263,7 @@
.thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("chunk error")));

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

ExecutionException exception = assertThrows(ExecutionException.class, future::get);
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
Expand All @@ -277,7 +277,7 @@
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done")));

TrackableStream stream = new TrackableStream("data");
callable.futureCall("resource-path", stream, null).get();
callable.futureCall("resource-path", stream).get();

assertThat(stream.closed).isTrue();
}
Expand All @@ -288,7 +288,7 @@
.thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("start failed")));

TrackableStream stream = new TrackableStream("data");
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream, null);
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream);
assertThrows(ExecutionException.class, future::get);

assertThat(stream.closed).isTrue();
Expand All @@ -306,7 +306,7 @@
});

TrackableStream stream = new TrackableStream("data");
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream, null);
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream);
assertThat(chunkStarted.await(5, TimeUnit.SECONDS)).isTrue();
future.cancel(true);

Expand All @@ -319,7 +319,7 @@
.thenThrow(new RuntimeException("sync start failure"));

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

ExecutionException exception = assertThrows(ExecutionException.class, future::get);
assertThat(exception.getCause()).isInstanceOf(RuntimeException.class);
Expand Down Expand Up @@ -348,7 +348,7 @@
ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "executor-done")));

ResumableUploadFuture<String> future =
customCallable.futureCall("resource-path", streamOf("data"), null);
customCallable.futureCall("resource-path", streamOf("data"));
assertThat(future.get()).isEqualTo("executor-done");
assertThat(tasksRun.get()).isGreaterThan(0);
} finally {
Expand All @@ -357,6 +357,98 @@
}
}

@Test
void testUploadCallable_convenienceOverload_noContext() throws Exception {
stubStartSession("https://upload.url/convenience");
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-conv")));

ResumableUploadFuture<String> future = callable.futureCall("resource-path", streamOf("data"));
assertThat(future.get()).isEqualTo("done-conv");
}

@Test
void testUploadCallable_withApiCallContext_mergesAndPassesContext() throws Exception {
stubStartSession("https://upload.url/context");
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-ctx")));

ApiCallContext customContext =
FakeCallContext.createDefault()
.withExtraHeaders(
java.util.Collections.singletonMap(
"X-Custom", java.util.Collections.singletonList("val")));
ResumableUploadFuture<String> future =
callable.futureCall("resource-path", streamOf("data"), customContext);
assertThat(future.get()).isEqualTo("done-ctx");

ArgumentCaptor<ApiCallContext> startContextCaptor =
ArgumentCaptor.forClass(ApiCallContext.class);
verify(mockStartCallable).futureCall(any(), startContextCaptor.capture());
assertThat(startContextCaptor.getValue()).isNotNull();
assertThat(((FakeCallContext) startContextCaptor.getValue()).getExtraHeaders())
.containsKey("X-Custom");

ArgumentCaptor<ApiCallContext> chunkContextCaptor =
ArgumentCaptor.forClass(ApiCallContext.class);
verify(mockChunkCallable).futureCall(any(), chunkContextCaptor.capture());
assertThat(chunkContextCaptor.getValue()).isNotNull();
assertThat(((FakeCallContext) chunkContextCaptor.getValue()).getExtraHeaders())
.doesNotContainKey("X-Custom");
}

@Test
void testUploadCallable_withSettings_mergesAndAppliesSettings() throws Exception {
stubStartSession("https://upload.url/settings");
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-settings")));

ResumableUploadCallSettings customSettings =
ResumableUploadCallSettings.newBuilder().setChunkSize(16).build();

ResumableUploadFuture<String> future =
callable.futureCall("resource-path", streamOf("data"), null, customSettings);
assertThat(future.get()).isEqualTo("done-settings");
}

@Test
void testUploadCallable_withSettingsConvenienceOverload_mergesAndAppliesSettings()
throws Exception {
stubStartSession("https://upload.url/settings-convenience");
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
.thenReturn(
ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-settings-conv")));

ResumableUploadCallSettings customSettings =
ResumableUploadCallSettings.newBuilder().setChunkSize(16).build();

ResumableUploadFuture<String> future =
callable.futureCall("resource-path", streamOf("data"), customSettings);
assertThat(future.get()).isEqualTo("done-settings-conv");
}

@Test
void testUploadCallable_withContextAndSettings_appliesBoth() throws Exception {
stubStartSession("https://upload.url/ctx-settings");
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-both")));

FakeCallContext customContext = FakeCallContext.createDefault();
ResumableUploadCallSettings customSettings =
ResumableUploadCallSettings.newBuilder().setChunkSize(16).build();

ResumableUploadFuture<String> future =
callable.futureCall("resource-path", streamOf("data"), customContext, customSettings);
assertThat(future.get()).isEqualTo("done-both");
}

@Test
void testResumeCall_throwsUnsupportedOperationException() {
assertThrows(

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

View check run for this annotation

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

Refactor the code of the lambda to have only one invocation possibly throwing a runtime exception. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaBu0MUVwtYwlDuGFbrx&open=AaBu0MUVwtYwlDuGFbrx&pullRequest=14251
UnsupportedOperationException.class,
() -> callable.resumeCall("https://upload.url/session", streamOf("data"), null));
}

private void stubStartSession(String uploadUrl) {
when(mockStartCallable.futureCall(any(), any()))
.thenReturn(
Expand Down
Loading