Skip to content

Commit c6c28b1

Browse files
whoweswhowes
committed
feat(gax): add rewindable stream buffer for chunk recovery
Introduce RewindableStreamBuffer managing a single-chunk buffer over an InputStream, supporting forward compaction and topping up upon recovery realignment without mark()/reset(). Enforces boundaries by throwing FailedPreconditionException when a server offset is below the base offset or beyond the current buffer window. Use payloadLength in ChunkUploadRequest to avoid allocating temporary byte arrays for full-sized chunks while reusing a single backing array.
1 parent 6b71690 commit c6c28b1

8 files changed

Lines changed: 652 additions & 33 deletions

File tree

‎sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
import java.io.ByteArrayInputStream;
4747
import java.io.InputStream;
4848
import java.nio.charset.StandardCharsets;
49+
import java.util.Arrays;
4950
import java.util.Collections;
5051
import java.util.List;
5152
import java.util.Map;
@@ -81,7 +82,12 @@ public Map<String, List<String>> getQueryParamNames(ChunkUploadRequest request)
8182

8283
@Override
8384
public byte[] getBinaryRequestBody(ChunkUploadRequest request) {
84-
return request.getPayload();
85+
int length = request.getPayloadLength();
86+
byte[] payload = request.getPayload();
87+
if (length == payload.length) {
88+
return payload;
89+
}
90+
return Arrays.copyOf(payload, length);
8591
}
8692

8793
@Override
@@ -110,7 +116,7 @@ private ResumableUploadChunkCallable(
110116
public ApiFuture<ChunkUploadResponse<ResponseT>> futureCall(
111117
ChunkUploadRequest request, @Nullable ApiCallContext inputContext) {
112118
Preconditions.checkNotNull(request);
113-
boolean isPayloadEmpty = request.getPayload().length == 0;
119+
boolean isPayloadEmpty = request.getPayloadLength() == 0;
114120
String command;
115121
if (request.isFinal()) {
116122
command = !isPayloadEmpty ? COMMAND_UPLOAD_FINALIZE : COMMAND_FINALIZE;

‎sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadRequest.java‎

Lines changed: 29 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import com.google.api.core.BetaApi;
3333
import com.google.api.core.InternalApi;
3434
import com.google.auto.value.AutoValue;
35+
import com.google.common.base.Preconditions;
3536
import org.jspecify.annotations.NullMarked;
3637

3738
/** Request value object for uploading a chunk to an active resumable upload session. */
@@ -48,6 +49,9 @@ public abstract class ChunkUploadRequest {
4849
@SuppressWarnings("mutable")
4950
public abstract byte[] getPayload();
5051

52+
/** The number of bytes within {@link #getPayload()} to upload. */
53+
public abstract int getPayloadLength();
54+
5155
/** The byte offset of this chunk in the overall stream. */
5256
public abstract long getOffset();
5357

@@ -56,8 +60,12 @@ public abstract class ChunkUploadRequest {
5660

5761
public abstract Builder toBuilder();
5862

63+
private static final int UNSET_PAYLOAD_LENGTH = Integer.MIN_VALUE;
64+
5965
public static Builder newBuilder() {
60-
return new AutoValue_ChunkUploadRequest.Builder().setFinal(false);
66+
return new AutoValue_ChunkUploadRequest.Builder()
67+
.setFinal(false)
68+
.setPayloadLength(UNSET_PAYLOAD_LENGTH);
6169
}
6270

6371
@AutoValue.Builder
@@ -66,10 +74,29 @@ public abstract static class Builder {
6674

6775
public abstract Builder setPayload(byte[] payload);
6876

77+
public abstract Builder setPayloadLength(int payloadLength);
78+
6979
public abstract Builder setOffset(long offset);
7080

7181
public abstract Builder setFinal(boolean isFinal);
7282

73-
public abstract ChunkUploadRequest build();
83+
abstract byte[] getPayload();
84+
85+
abstract int getPayloadLength();
86+
87+
abstract ChunkUploadRequest autoBuild();
88+
89+
public ChunkUploadRequest build() {
90+
if (getPayloadLength() == UNSET_PAYLOAD_LENGTH) {
91+
setPayloadLength(getPayload().length);
92+
}
93+
ChunkUploadRequest request = autoBuild();
94+
Preconditions.checkArgument(
95+
request.getPayloadLength() >= 0, "payloadLength must be non-negative");
96+
Preconditions.checkArgument(
97+
request.getPayloadLength() <= request.getPayload().length,
98+
"payloadLength exceeds payload array length");
99+
return request;
100+
}
74101
}
75102
}

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

Lines changed: 10 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -43,12 +43,10 @@
4343
import com.google.api.gax.retrying.RetrySettings;
4444
import com.google.api.gax.retrying.RetryingFuture;
4545
import com.google.api.gax.retrying.ScheduledRetryingExecutor;
46-
import com.google.common.io.ByteStreams;
4746
import com.google.common.util.concurrent.MoreExecutors;
4847
import java.io.IOException;
4948
import java.io.InputStream;
5049
import java.time.Duration;
51-
import java.util.Arrays;
5250
import java.util.concurrent.CancellationException;
5351
import java.util.concurrent.atomic.AtomicBoolean;
5452
import java.util.concurrent.atomic.AtomicLong;
@@ -74,17 +72,13 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
7472
.setMaxAttempts(5)
7573
.build();
7674

77-
private static final byte[] EMPTY_PAYLOAD = new byte[0];
78-
7975
private final AtomicBoolean dispatching = new AtomicBoolean(false);
8076
private final AtomicLong nextChunkOffset = new AtomicLong(-1L);
8177

8278
private final RetryingCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>>
8379
retryingChunkCallable;
8480
private final String uploadUrl;
85-
private final InputStream payload;
86-
private final byte[] buffer;
87-
private final int chunkSize;
81+
private final RewindableStreamBuffer buffer;
8882
private final ApiCallContext callContext;
8983
private final SettableApiFuture<ResponseT> result = SettableApiFuture.create();
9084

@@ -97,11 +91,10 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
9791
ClientContext clientContext) {
9892
checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
9993
this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null");
100-
this.payload = checkNotNull(payload, "payload must not be null");
101-
this.chunkSize = chunkSize;
94+
checkNotNull(payload, "payload must not be null");
10295
this.callContext = checkNotNull(callContext, "callContext must not be null");
10396
checkNotNull(clientContext, "clientContext must not be null");
104-
this.buffer = new byte[chunkSize];
97+
this.buffer = new RewindableStreamBuffer(payload, chunkSize, uploadUrl);
10598

10699
RetryAlgorithm<ChunkUploadResponse<ResponseT>> retryAlgorithm =
107100
new RetryAlgorithm<>(
@@ -141,33 +134,24 @@ private void transmitSingleChunk(long currentOffset) {
141134
return;
142135
}
143136

144-
int bytesRead;
145137
try {
146-
bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize);
138+
buffer.fill(currentOffset);
147139
} catch (IOException e) {
148140
result.setException(e);
149141
return;
150142
}
151143

152-
boolean isFinal = bytesRead < chunkSize;
153-
byte[] chunkPayload;
154-
if (bytesRead == chunkSize) {
155-
chunkPayload = buffer;
156-
} else if (bytesRead == 0) {
157-
chunkPayload = EMPTY_PAYLOAD;
158-
} else {
159-
chunkPayload = Arrays.copyOf(buffer, bytesRead);
160-
}
161-
162144
ChunkUploadRequest chunkRequest =
163145
ChunkUploadRequest.newBuilder()
164146
.setUploadUrl(uploadUrl)
165-
.setPayload(chunkPayload)
166-
.setOffset(currentOffset)
167-
.setFinal(isFinal)
147+
.setPayload(buffer.getBuffer())
148+
.setPayloadLength(buffer.getPayloadLength())
149+
.setOffset(buffer.getBufferBaseOffset())
150+
.setFinal(buffer.isFinal())
168151
.build();
169152

170-
long chunkLength = chunkPayload.length;
153+
long chunkLength = chunkRequest.getPayloadLength();
154+
boolean isFinal = chunkRequest.isFinal();
171155
try {
172156
RetryingFuture<ChunkUploadResponse<ResponseT>> retryingFuture =
173157
retryingChunkCallable.futureCall(chunkRequest, callContext);

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

Lines changed: 157 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,157 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Redistribution and use in source and binary forms, with or without
5+
* modification, are permitted provided that the following conditions are
6+
* met:
7+
*
8+
* * Redistributions of source code must retain the above copyright
9+
* notice, this list of conditions and the following disclaimer.
10+
* * Redistributions in binary form must reproduce the above
11+
* copyright notice, this list of conditions and the following disclaimer
12+
* in the documentation and/or other materials provided with the
13+
* distribution.
14+
* * Neither the name of Google LLC nor the names of its
15+
* contributors may be used to endorse or promote products derived from
16+
* this software without specific prior written permission.
17+
*
18+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19+
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20+
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
21+
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
22+
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
23+
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
24+
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
25+
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
26+
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27+
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
28+
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29+
*/
30+
package com.google.api.gax.rpc;
31+
32+
import static com.google.common.base.Preconditions.checkArgument;
33+
import static com.google.common.base.Preconditions.checkNotNull;
34+
35+
import com.google.common.io.ByteStreams;
36+
import java.io.IOException;
37+
import java.io.InputStream;
38+
import org.jspecify.annotations.NullMarked;
39+
40+
/**
41+
* Manages a single-chunk buffer over an {@link InputStream} for resumable uploads.
42+
*
43+
* <p>The buffer holds at most one chunk of data in a reused backing array. It supports forward
44+
* compaction and topping up upon recovery realignment, and enforces the boundary condition that
45+
* requests to rewind before the buffer's base offset fail with an unrecoverable {@link
46+
* FailedPreconditionException}.
47+
*/
48+
@NullMarked
49+
final class RewindableStreamBuffer {
50+
51+
private final InputStream inputStream;
52+
private final int chunkSize;
53+
private final String uploadUrl;
54+
private final byte[] buffer;
55+
56+
private long bufferBaseOffset;
57+
private int payloadLength;
58+
private boolean isFinal;
59+
private boolean streamExhausted;
60+
61+
RewindableStreamBuffer(InputStream inputStream, int chunkSize, String uploadUrl) {
62+
this.inputStream = checkNotNull(inputStream, "inputStream must not be null");
63+
checkArgument(chunkSize > 0, "chunkSize must be > 0");
64+
this.chunkSize = chunkSize;
65+
this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null");
66+
this.buffer = new byte[chunkSize];
67+
this.bufferBaseOffset = 0L;
68+
this.payloadLength = 0;
69+
this.isFinal = false;
70+
this.streamExhausted = false;
71+
}
72+
73+
/**
74+
* Advances the buffer from the stream starting at {@code targetOffset}, reading up to chunk size.
75+
*
76+
* @param targetOffset the absolute stream offset corresponding to the start of this chunk
77+
* @throws IOException if reading from the stream fails
78+
*/
79+
void fill(long targetOffset) throws IOException {
80+
this.bufferBaseOffset = targetOffset;
81+
this.payloadLength = ByteStreams.read(inputStream, buffer, 0, chunkSize);
82+
this.isFinal = (payloadLength < chunkSize);
83+
if (this.isFinal) {
84+
this.streamExhausted = true;
85+
}
86+
}
87+
88+
/**
89+
* Realigns the buffer window to {@code committedOffset}.
90+
*
91+
* <p>Compacts forward within the existing buffer to discard already-committed bytes, and then
92+
* tops up the buffer to capacity from the underlying stream.
93+
*
94+
* @param committedOffset the server's committed byte offset
95+
* @throws FailedPreconditionException if {@code committedOffset} is below the buffer's base
96+
* offset or beyond the current buffer window
97+
* @throws IOException if reading from the stream fails
98+
*/
99+
void realignTo(long committedOffset) throws IOException {
100+
if (committedOffset < bufferBaseOffset) {
101+
throw UploadErrors.protocolViolation(
102+
String.format(
103+
"Server committed offset %d is below buffer base offset %d for upload URL %s; cannot"
104+
+ " rewind stream before buffer base",
105+
committedOffset, bufferBaseOffset, uploadUrl));
106+
}
107+
108+
if (committedOffset > bufferBaseOffset + payloadLength) {
109+
throw UploadErrors.protocolViolation(
110+
String.format(
111+
"Server committed offset %d is beyond current buffer window [%d, %d] for upload URL"
112+
+ " %s",
113+
committedOffset, bufferBaseOffset, bufferBaseOffset + payloadLength, uploadUrl));
114+
}
115+
116+
int committedWithinBuffer = (int) (committedOffset - bufferBaseOffset);
117+
int remainingBytes = payloadLength - committedWithinBuffer;
118+
119+
if (remainingBytes > 0 && committedWithinBuffer > 0) {
120+
System.arraycopy(buffer, committedWithinBuffer, buffer, 0, remainingBytes);
121+
}
122+
123+
this.bufferBaseOffset = committedOffset;
124+
this.payloadLength = remainingBytes;
125+
126+
if (!streamExhausted && payloadLength < chunkSize) {
127+
int space = chunkSize - payloadLength;
128+
int additionalRead = ByteStreams.read(inputStream, buffer, payloadLength, space);
129+
payloadLength += additionalRead;
130+
if (additionalRead < space) {
131+
streamExhausted = true;
132+
}
133+
}
134+
135+
this.isFinal = streamExhausted;
136+
}
137+
138+
byte[] getBuffer() {
139+
return buffer;
140+
}
141+
142+
int getPayloadLength() {
143+
return payloadLength;
144+
}
145+
146+
long getBufferBaseOffset() {
147+
return bufferBaseOffset;
148+
}
149+
150+
boolean isFinal() {
151+
return isFinal;
152+
}
153+
154+
boolean isEmpty() {
155+
return payloadLength == 0;
156+
}
157+
}

0 commit comments

Comments
 (0)