Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces global timeout handling for resumable uploads in ResumableUploadChunkCoordinator using a scheduled executor to trigger a DeadlineExceededException if the upload exceeds the specified duration. It also includes comprehensive unit tests to verify timeout behavior, clean cancellations, and payload stream closure. The review feedback highlights two key improvements: ensuring thread safety by reading uploadSessionUrl within a synchronized block in onTimeout(), and preventing potential test hangs by specifying timeouts on future.get() calls in the newly added unit tests.
| private void onTimeout() { | ||
| synchronized (lock) { | ||
| if (done) { | ||
| return; | ||
| } | ||
| } | ||
| String message = | ||
| uploadSessionUrl != null | ||
| ? "Resumable upload timed out for session: " + uploadSessionUrl | ||
| : "Resumable upload timed out before session initiation completed"; | ||
| finish(null, new DeadlineExceededException(message, null, TIMEOUT_STATUS_CODE, false)); | ||
| } |
There was a problem hiding this comment.
The field uploadSessionUrl is read outside of the synchronized (lock) block. If uploadSessionUrl is not declared volatile or is guarded by lock, this can lead to a data race and memory visibility issues across threads (e.g., the timeout thread might see a stale null value even after the session URL has been initialized). To ensure thread safety and consistent visibility, read uploadSessionUrl inside the synchronized (lock) block.
private void onTimeout() {
String sessionUrl;
synchronized (lock) {
if (done) {
return;
}
sessionUrl = uploadSessionUrl;
}
String message =
sessionUrl != null
? "Resumable upload timed out for session: " + sessionUrl
: "Resumable upload timed out before session initiation completed";
finish(null, new DeadlineExceededException(message, null, TIMEOUT_STATUS_CODE, false));
}| ResumableUploadFuture<String> future = | ||
| callable.futureCall("resource-path", streamOf("hello"), timeoutSettings); | ||
|
|
||
| ExecutionException exception = assertThrows(ExecutionException.class, future::get); |
There was a problem hiding this comment.
Calling future.get() without a timeout in a timeout-related unit test can cause the test suite to hang indefinitely if the timeout mechanism fails to fire. It is a best practice to always specify a timeout (e.g., 5 seconds) when waiting on futures in tests to prevent blocking the CI/CD pipeline.
| ExecutionException exception = assertThrows(ExecutionException.class, future::get); | |
| ExecutionException exception = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); |
3011b46 to
3e2fdd4
Compare
3e2fdd4 to
4dfdd22
Compare
4dfdd22 to
3794533
Compare
497d777 to
a5ffaf3
Compare
a5ffaf3 to
fc40846
Compare
Enforces ResumableUploadCallSettings.getGlobalTimeout() in ResumableUploadChunkCoordinator across the upload lifecycle. Cancels in-flight RPCs and completes the future with DeadlineExceededException when the deadline is exceeded.
fc40846 to
6960f29
Compare
|
|
1 participant
Enforces
ResumableUploadCallSettings.getGlobalTimeout()inResumableUploadChunkCoordinatoracross the upload lifecycle. Cancels in-flight RPCs and completes the future withDeadlineExceededExceptionwhen the deadline is exceeded.