Repository navigation
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 implements session resumption for resumable uploads in the GAX library, allowing uploads to resume from a saved session URL by querying the server's committed offset and fast-forwarding the stream. The review feedback identifies a critical bug in ResumableUploadChunkCoordinator where a transient failure during the initial resume query causes retries to incorrectly bypass the stream fast-forwarding step, resulting in an IllegalStateException. To fix this, the reviewer suggests tracking the initial resume state using a private member variable instead of a method parameter. Additionally, the reviewer flags the accidental deletion of the compliance_suite.json file, which should be restored.
| void start() { | ||
| progressTracker.onStarted(uploadUrl, chunkSize); | ||
| registerCancellationListener(); | ||
| try { | ||
| buffer.fill(); | ||
| transmitChunk(); | ||
| } catch (Throwable t) { | ||
| uploadResultFuture.setException(t); | ||
| } | ||
| } | ||
|
|
||
| void resume() { | ||
| registerCancellationListener(); | ||
| queryStatus(/* isInitialResume= */ true); | ||
| } |
There was a problem hiding this comment.
If the initial resume query fails due to a transient network error, the scheduled retry will call queryStatus(false). Because isInitialResume is hardcoded to false in the retry, the retried query will be treated as a normal recovery query. When it succeeds, handleQueryResponse will call buffer.recoverAt(committedOffset) instead of buffer.resumeAt(committedOffset). Since buffer.resumeAt was never called, the stream was never skipped to the committed offset, and buffer.recoverAt will throw an IllegalStateException because the committed offset is beyond the uninitialized buffer window.\n\nTo fix this, track the initial resume state using a private field isInitialResume in ResumableUploadChunkCoordinator instead of passing it as a parameter. This field is set to true in resume(), and set to false inside handleQueryResponse after a successful buffer.resumeAt call.
private boolean isInitialResume;\n\n void start() {\n progressTracker.onStarted(uploadUrl, chunkSize);\n registerCancellationListener();\n try {\n buffer.fill();\n transmitChunk();\n } catch (Throwable t) {\n uploadResultFuture.setException(t);\n }\n }\n\n void resume() {\n isInitialResume = true;\n registerCancellationListener();\n queryStatus();\n }| tryRegisterInFlightFuture( | ||
| executor.schedule( | ||
| this::queryStatus, | ||
| () -> queryStatus(/* isInitialResume= */ false), | ||
| recoverySettings.getRandomizedRetryDelayDuration().toNanos(), | ||
| TimeUnit.NANOSECONDS)); | ||
| } catch (Throwable t) { | ||
| uploadResultFuture.setException(t); | ||
| } | ||
| } | ||
|
|
||
| private void queryStatus() { | ||
| private void queryStatus(boolean isInitialResume) { |
There was a problem hiding this comment.
Revert the signature of queryStatus and the retry scheduling to use the original parameterless queryStatus() method, leveraging the new isInitialResume field instead.
tryRegisterInFlightFuture(\n executor.schedule(\n this::queryStatus,\n recoverySettings.getRandomizedRetryDelayDuration().toNanos(),\n TimeUnit.NANOSECONDS));\n } catch (Throwable t) {\n uploadResultFuture.setException(t);\n }\n }\n\n private void queryStatus() {| } | ||
| try { | ||
| handleQueryResponse(queryResponse); | ||
| handleQueryResponse(queryResponse, isInitialResume); |
There was a problem hiding this comment.
Revert the call to handleQueryResponse to use the original parameterless signature.
| handleQueryResponse(queryResponse, isInitialResume); | |
| handleQueryResponse(queryResponse); |
| private void handleQueryResponse( | ||
| QueryStatusResponse<ResponseT> queryResponse, boolean isInitialResume) throws IOException { |
There was a problem hiding this comment.
Revert the signature of handleQueryResponse to the original parameterless version.
private void handleQueryResponse(QueryStatusResponse<ResponseT> queryResponse)\n throws IOException {| if (isInitialResume) { | ||
| buffer.resumeAt(committedOffset); | ||
| } else { | ||
| buffer.recoverAt(committedOffset); | ||
| } |
There was a problem hiding this comment.
Update handleQueryResponse to use the isInitialResume field and set it to false after successfully calling buffer.resumeAt.
if (isInitialResume) {\n isInitialResume = false;\n buffer.resumeAt(committedOffset);\n } else {\n buffer.recoverAt(committedOffset);\n }| @@ -1,240 +0,0 @@ | |||
| { | |||
There was a problem hiding this comment.
The deletion of compliance_suite.json appears to be accidental as it is unrelated to the resumable upload session resumption feature. If this was unintentional, please restore the file.
|
|
This branch has not been deployed
1 participant
Implements
ResumableUploadCallable.resumeCall(...)to support resuming interrupted resumable upload sessions from a saved session URL and effective chunk size.