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

This is needed to meet the resumable upload requirement to allow custom
headers and other settings to be applied on a per-call basis.

Author: whowes

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java

@@ -48,19 +48,41 @@ public abstract class ResumableUploadCallable<RequestT, ResponseT> {
   protected ResumableUploadCallable() {}
 
   /**
-   * Performs a new resumable upload asynchronously.
+   * 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 call settings 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 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);
 
   /**
    * Resumes an existing resumable upload session asynchronously using a saved session URL.

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java

@@ -70,15 +70,18 @@ public ResumableUploadCallableImpl(
 
   @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 = clientContext.getDefaultCallContext().merge(context);
 
     ApiFuture<ResumableUploadSession> startFuture;
     try {
-      startFuture =
-          client.startUploadCallable().futureCall(request, clientContext.getDefaultCallContext());
+      startFuture = client.startUploadCallable().futureCall(request, effectiveCallContext);
     } catch (Throwable t) {
       startFuture = ApiFutures.immediateFailedFuture(t);
     }

sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java

@@ -313,6 +313,87 @@ void testUploadCallable_closesPayloadOnStartSyncFailure() {
     assertThat(stream.closed).isTrue();
   }
 
+  @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, null);
+    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_withSettings_delegatesWithNullContext() 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(
+        UnsupportedOperationException.class,
+        () -> callable.resumeCall("https://upload.url/session", streamOf("data"), null));
+  }
+
   private void stubStartSession(String uploadUrl) {
     when(mockStartCallable.futureCall(any(), any()))
         .thenReturn(