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

Add 2-, 3-, and 4-argument overloads to ResumableUploadCallable for
futureCall and resumeCall, supporting:
1. Default context and settings (used by generated ServiceClient convenience methods).
2. Per-request ApiCallContext overrides for transport metadata (extra headers, credentials).
3. Per-request ResumableUploadCallSettings overrides for state-machine knobs per go/sdk:java-scotty-design.
4. Full method accepting both call context and settings overrides.

ResumableUploadCallableImpl performs a 3-tier precedence merge on settings
and merges ApiCallContext.

Author: whowes

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

@@ -48,19 +48,57 @@ 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 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

@@ -75,14 +75,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 = defaultCallContext.merge(context);
 
     ApiFuture<ResumableUploadSession> startFuture;
     try {
-      startFuture = client.startUploadCallable().futureCall(request, defaultCallContext);
+      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

@@ -357,6 +357,82 @@ public void execute(Runnable command) {
     }
   }
 
+  @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_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(