feat(gax): implement baseline Callable and Future for resumable uploads

[whowes]

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

[whowes]

/gemini review

[whowes]

/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]

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]

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

[whowes]

Done.

[blakeli0]

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

[whowes]

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.

[blakeli0]

I think there are two issues here:

1. Should we take the responsibility of closing the stream? Usually whoever creates the stream is responsible for it.
2. If we do want to take the responsibility, using try-with-resources is preferred than manually closing it.

[whowes]

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:

1. 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.

2. 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.

[blakeli0]

I think the whole upload(including the initial call) can be done in a single thread. Using transformAsync may transform the future in a different thread, we can ended up with a lot of thread when uploading large files.

[whowes]

I'm not a big fan of pinning each upload to a thread (making the entire operation basically synchronous on that thread). The biggest issue that I see is that then a thread belonging to the provided Executor is occupied for the entire duration of potentially very long uploads and unavailable for other tasks, when a lot of the time the upload thread would be doing no work (waiting for the response to come back from the server).

It seems that the Executor passed down here with typical GAX callable defaults would be fixed thread pools, in which case keeping an upload pinned to a thread would actually be particularly problematic. So unless the Executor we use is specifically dedicated to uploads (which IIUC it wouldn't be unless we did something pretty atypical for GAX callables) that could also interfere with other library operations.

[blakeli0]

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.

[whowes]

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]

In general, I think the current structure of the how we make initial call and upload call can be improved. The nested calls of ApiFutures.transformAsync is not easy to read, and may have performance concerns. Some future calls can be made in the callable as well. There could also be a wrapper callable/future that does the whole uploading.

A pseudo code I'm thinking in the Callable is

StartUploadFuture startUploadFuture = client.startUploadCallable().futureCall();
UploadWholeCallable uploadWholeCallable = new UploadWholeCallable(startUploadFuture, client);
UploadWholeFuture uploadWholeFuture = uploadWholeCallable().futureCall();

return new ResumableUploadFuture(startUploadFuture, uploadWholeFuture).

This is similar to OperationCallableImpl.

Let me know what you think and if I missed anything.

[whowes]

I think moving away from the transformAsync chain is a good move, in particular because I think the structure would become tricky to extend with the failure->query->upload recovery loop that we'll be adding soon.

With my latest changes I've spread out the responsibilities across multiple parties kind of similarly to your proposal:

• ResumableUploadCallableImpl initiates the startUpload and hands it off to ResumableUploadFutureImpl
• ResumableUploadFutureImpl adds a callback on the start future that will hand off to ResumableUploadChunkCoordinator when the upload URL is available. It keeps track of which operation (start or upload) is in flight so that cancellations can get percolated down appropriately (so it's basically the "whole future" - I didn't really see why there needed to be an additional layer of wrapping)
• ResumableUploadChunkCoordinator is kind of the role of the "UploadWholeCallable" (though it's not actually a callable) and manages the chunk uploading

In future PRs (post generator work) I envision new responsibilities being balanced this way:

• Retry of start command managed within ResumableUploadCallableImpl (via standard callable wrapping)
• Overall session timeout and status/progress listeners managed by ResumableUploadFutureImpl
• Chunking retries and query/upload recovery loop managed inside ResumableUploadChunkCoordinator

WDYT?

[sonarqubecloud]

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

Issues
9 New issues
0 Accepted issues

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

See analysis details on SonarQube Cloud

[sonarqubecloud]

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

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

See analysis details on SonarQube Cloud