feat(gax): implement baseline Callable and Future for resumable uploads by whowes · Pull Request #14241 · googleapis/google-cloud-java

whowes · 2026-09-02T17:02:40Z

This implementation supports the happy path only; retries, recovery, timeouts, per-call settings, and progress tracking will be added in subsequent phases.

whowes · 2026-09-02T21:55:10Z

/gemini review

whowes · 2026-09-02T22:38:06Z

/gemini review

gemini-code-assist

Code Review

This pull request introduces the concrete implementation of ResumableUploadCallable and ResumableUploadFuture (ResumableUploadCallableImpl and ResumableUploadFutureImpl) to coordinate resumable upload sessions and stream chunks asynchronously, along with comprehensive unit tests. The feedback highlights a potential issue where performing blocking I/O (ByteStreams.read) inside asynchronous future callbacks could lead to thread starvation or deadlocks if a limited executor is used, suggesting either documenting executor requirements or offloading the blocking read.

gemini-code-assist · 2026-09-02T22:40:38Z

+    byte[] buffer = new byte[chunkSize];
+    int bytesRead;
+    try {
+      bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize);


Performing blocking I/O (ByteStreams.read) inside asynchronous future callbacks (which run on the provided executor) can lead to thread starvation or deadlocks if the executor is a direct executor or a limited thread pool (such as gRPC network threads). Consider documenting that the executor passed to the callable must be a dedicated thread pool suitable for blocking I/O operations, or offloading the blocking read to a dedicated I/O executor.

blakeli0 · 2026-09-03T21:30:27Z

+    ResumableUploadCallSettings effectiveSettings = defaultCallSettings.merge(settings);
+
+    return ResumableUploadFutureImpl.create(
+        client, request, payload, effectiveSettings.getChunkSize(), defaultCallContext, executor);


Since we know there will be more configurations, can we pass the whole settings class to the future?

blakeli0 · 2026-09-03T21:55:26Z

+                        + " incomplete status"));
+          }
+          // Continuation: asynchronously transmit subsequent chunk with updated offset.
+          return transmitChunks(


Can we use a while loop instead of recursive calls? There is always stackoverflow concerns using recursives.

IIUC a conventional while would pretty much map to the single thread per upload idea (and keeping the thread pinned to the upload even while it's blocking on I/O)? That seems problematic to me (see other comment).

On the recursion point (this particular code is restructured, but the new code chains kinda similarly): it looks recursion-ish, but stack frames don't actually accumulate. futureCall returns immediately, transmitChunk returns, and the transmitChunk stack frame is popped. When the chunk future finishes later, the executor invokes the callback and it's not coupled with the stack frame of when it was scheduled. (The overflow miiight be a risk if the executor used here was a DirectExecutor which was possible in the last snapshot, but I switched to a ScheduledExecutorService as I think we'll need that when we layer in retries.)

IIUC this callback chaining pattern is pretty similar to CallbackChainRetryingFuture which does a loop inside its completion listener (submit -> setAttemptFuture -> attach listener -> repeat) across retry attempts rather than blocking a thread in a loop.

SGTM. I agree that there are no recursive concerns also.

My original idea is to use an IO thread (which has a large thread pool) for the whole operation since each low-level call is done in a separate thread anyway, this can simplify certain implementation such as updating/reading upload url, reusing the same byte[] etc.. But this requires us to have a "main thread" for each upload session that monitors the progress of whole operation, which still poses risks of thread starvation if we have a lot of uploads.

blakeli0 · 2026-09-03T22:15:17Z

+                return transmitChunks(client, payload, chunkSize, callContext, url, 0L, executor);
+              },
+              executor);
+      sessionFuture.addListener(() -> closePayload(payload), executor);


I think there are two issues here:

Should we take the responsibility of closing the stream? Usually whoever creates the stream is responsible for it.

If we do want to take the responsibility, using try-with-resources is preferred than manually closing it.

Typically the caller that provides the resource is responsible for closing it in synchronous code, but it's problematic in async calls when using try-with-resources:

try (InputStream stream = getStream()) { callable.futureCall(request, stream); } // <-- stream closed future.get(); <-- possible failure if callable tried to use stream when closed

This is an issue for both 1 and 2 IMO:

with the typical convenient approach not so safe and easy for async it's on the caller to figure out when it's safe to close the stream, which can be tricky to keep track of particularly if the result is accessed far away (in code) from where the stream was created and passed along. So having responsibility transferred to the async task - provided that fact is clearly documented - removes that cognitive load from the caller.

On the future impl side if we are making our impl mostly asynchronous rather than writing a synchronous while-style with a dedicated thread per upload (which IMO we should do, the theme of several of my other comments :) then we suffer from the same problem of simple try-with-resources closing the streams prematurely.

Another idea is to change the input from a raw stream to a supplier of stream. So that Gax takes the responsibility of opening and closing the stream completely.
That being said, it does not block this PR and can be refactored later.

That makes sense - possibly this could be additive and the existing version that takes an InputStream directly could then be retrofit to use a supplier that returns the provided instance (requiring all callers to use a supplier may be a bit inconvenient for non-power users). Let's revisit this later.

blakeli0 · 2026-09-03T22:24:10Z

+  private static final byte[] EMPTY_PAYLOAD = new byte[0];
+
+  private final InputStream payload;
+  private final AtomicReference<@Nullable String> uploadSessionUrl;


ResumableUploadFuture represents one main upload session and there should be only one thread modifying this url. I don't think we need to use AtomicReference.

True, there is only one writer so the AtomicReference is probably overkill (though I don't think it adds much overhead). I switched to a regular @Nullable field but I believe it needs to be volatile since multiple threads may read it (which could happen via getUploadSessionUrl()).

blakeli0 · 2026-09-09T06:16:07Z

+                        + " incomplete status"));
+          }
+          // Continuation: asynchronously transmit subsequent chunk with updated offset.
+          return transmitChunks(


SGTM. I agree that there are no recursive concerns also.

My original idea is to use an IO thread (which has a large thread pool) for the whole operation since each low-level call is done in a separate thread anyway, this can simplify certain implementation such as updating/reading upload url, reusing the same byte[] etc.. But this requires us to have a "main thread" for each upload session that monitors the progress of whole operation, which still poses risks of thread starvation if we have a lot of uploads.

blakeli0 · 2026-09-09T22:07:09Z

+                return transmitChunks(client, payload, chunkSize, callContext, url, 0L, executor);
+              },
+              executor);
+      sessionFuture.addListener(() -> closePayload(payload), executor);


Another idea is to change the input from a raw stream to a supplier of stream. So that Gax takes the responsibility of opening and closing the stream completely.
That being said, it does not block this PR and can be refactored later.

blakeli0 · 2026-09-09T22:12:23Z

+    if (inFlight != null) {
+      inFlight.cancel(mayInterruptIfRunning);
+    }
+    closePayload();


For follow up PRs, I think we need to give it more thought how to close the stream in cancel. Because the stream might be processing in another chunk uploading thread.

Acknowledged.

blakeli0 · 2026-09-09T22:13:45Z

+    try {
+      payload.close();
+    } catch (IOException ignored) {
+      // Suppressed during stream cleanup


For follow up PRs, we should think about how to handle this error case. I don't think we should ignore it.

Acknowledged.

…uture

sonarqubecloud · 2026-09-10T01:22:46Z

Quality Gate passed for 'gapic-generator-java-root'

Issues
9 New issues
0 Accepted issues

Measures
0 Security Hotspots
86.1% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

sonarqubecloud · 2026-09-10T01:24:14Z

Quality Gate failed for 'gapic-generator-java-root'

Failed conditions
0.0% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube Cloud

…JsonCallableFactory (#14242) This change wires up the minimal Callable implementation from #14241 into the relevant existing factory classes.

This comment was marked as outdated.

Sign in to view

whowes mentioned this pull request Sep 2, 2026

feat(gax): add ResumableUploadCallable creation to Callables and HttpJsonCallableFactory #14242

Merged

whowes force-pushed the whowes/resumable-upload-happy-path branch 5 times, most recently from 554caca to b8fc38f Compare September 2, 2026 20:00

This comment was marked as outdated.

Sign in to view

whowes force-pushed the whowes/resumable-upload-happy-path branch 2 times, most recently from 234e79f to 6845669 Compare September 2, 2026 22:32

gemini-code-assist Bot reviewed Sep 2, 2026

View reviewed changes

whowes requested a review from blakeli0 September 2, 2026 22:59

whowes marked this pull request as ready for review September 2, 2026 22:59

whowes requested review from a team as code owners September 2, 2026 22:59

whowes force-pushed the whowes/resumable-upload-happy-path branch from 6845669 to 61fd370 Compare September 3, 2026 05:29

blakeli0 reviewed Sep 3, 2026

View reviewed changes

whowes force-pushed the whowes/resumable-upload-happy-path branch 8 times, most recently from 511bdb1 to 1ac786d Compare September 8, 2026 22:15

whowes requested a review from blakeli0 September 9, 2026 00:27

blakeli0 reviewed Sep 9, 2026

View reviewed changes

whowes force-pushed the whowes/resumable-upload-happy-path branch from 1ac786d to 3472f61 Compare September 9, 2026 15:50

whowes requested a review from blakeli0 September 9, 2026 16:20

whowes force-pushed the whowes/resumable-upload-happy-path branch from 3472f61 to 3c3310d Compare September 9, 2026 20:15

blakeli0 approved these changes Sep 9, 2026

View reviewed changes

feat(gax): implement MVP ResumableUploadCallable and ResumableUploadF…

353a559

…uture

whowes force-pushed the whowes/resumable-upload-happy-path branch from 3c3310d to 353a559 Compare September 10, 2026 01:14

whowes added the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Sep 10, 2026

yoshi-kokoro removed the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Sep 10, 2026

whowes merged commit 5a54db9 into main Sep 10, 2026
300 of 306 checks passed

whowes deleted the whowes/resumable-upload-happy-path branch September 10, 2026 18:34

Uh oh!

Conversation

whowes commented Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

This comment was marked as outdated.

Uh oh!

whowes commented Sep 2, 2026

Uh oh!

This comment was marked as outdated.

Uh oh!

whowes commented Sep 2, 2026

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Sep 2, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sonarqubecloud Bot commented Sep 10, 2026

Quality Gate passed for 'gapic-generator-java-root'

Uh oh!

sonarqubecloud Bot commented Sep 10, 2026

Quality Gate failed for 'gapic-generator-java-root'

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

whowes commented Sep 2, 2026 •

edited

Loading