Skip to content

Commit 461fc12

Browse files
whoweswhowes
committed
feat(gax): standardize mapping terminal resumable upload exceptions to ApiException
1 parent 0c481bd commit 461fc12

5 files changed

Lines changed: 148 additions & 65 deletions

File tree

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

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
9999
this.executor = checkNotNull(executor, "executor must not be null");
100100
this.recoverySettings = recoveryAlgorithm.createFirstAttempt();
101101
this.progressTracker = checkNotNull(progressTracker, "progressTracker must not be null");
102-
this.buffer = new RewindableStreamBuffer(payload, chunkSize, uploadUrl);
102+
this.buffer = new RewindableStreamBuffer(payload, chunkSize);
103103
}
104104

105105
ApiFuture<ResponseT> getFuture() {
@@ -150,8 +150,7 @@ public void onSuccess(ChunkUploadResponse<ResponseT> response) {
150150
if (response.getUploadStatus() == ResumableUploadStatus.UNKNOWN) {
151151
recover(
152152
new IllegalStateException(
153-
"Chunk upload response missing X-Goog-Upload-Status header for upload URL: "
154-
+ uploadUrl));
153+
"Chunk upload response missing X-Goog-Upload-Status header"));
155154
} else {
156155
try {
157156
handleChunkResponse(response);
@@ -263,8 +262,7 @@ private boolean tryRegisterInFlightFuture(Future<?> future) {
263262
private void handleQueryResponse(QueryStatusResponse<ResponseT> queryResponse)
264263
throws IOException {
265264
if (queryResponse.getUploadStatus() == ResumableUploadStatus.UNKNOWN) {
266-
throw new IllegalStateException(
267-
"Query status response missing X-Goog-Upload-Status header for upload URL: " + uploadUrl);
265+
throw new IllegalStateException("Query status response missing X-Goog-Upload-Status header");
268266
}
269267
if (queryResponse.getUploadStatus() == ResumableUploadStatus.FINAL) {
270268
progressTracker.onFinalized(buffer.getBufferBaseOffset() + buffer.getPayload().length);
@@ -274,8 +272,7 @@ private void handleQueryResponse(QueryStatusResponse<ResponseT> queryResponse)
274272
Long committedOffset = queryResponse.getCommittedOffset();
275273
if (committedOffset == null) {
276274
throw new IllegalStateException(
277-
"Incomplete query status response did not include a committed offset for upload URL: "
278-
+ uploadUrl);
275+
"Incomplete query status response did not include a committed offset");
279276
}
280277
progressTracker.onOffsetReceived(committedOffset);
281278
buffer.realignTo(committedOffset);
@@ -290,8 +287,7 @@ private void handleChunkResponse(ChunkUploadResponse<ResponseT> response) throws
290287
uploadResultFuture.setException(
291288
new IllegalStateException(
292289
"Upload stream ended and final chunk was transmitted, but server returned"
293-
+ " incomplete status for upload URL: "
294-
+ uploadUrl));
290+
+ " incomplete status"));
295291
} else {
296292
madeProgressSinceRecovery = true;
297293
buffer.fill();

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

Lines changed: 49 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -212,14 +212,15 @@ public void onFailure(Throwable t) {
212212
}
213213

214214
private void onTimeout() {
215-
String sessionUrl = uploadSessionUrl;
216215
String message;
217-
if (sessionUrl != null) {
218-
message = "Resumable upload timed out for session: " + sessionUrl;
216+
if (uploadSessionUrl != null) {
217+
message = "Resumable upload timed out";
219218
} else {
220219
message = "Resumable upload timed out before session initiation completed";
221220
}
222-
fail(ApiExceptionFactory.createException(message, null, TIMEOUT_STATUS_CODE, false));
221+
fail(
222+
ApiExceptionFactory.createException(
223+
message, null, statusCodeOf(StatusCode.Code.DEADLINE_EXCEEDED), false));
223224
}
224225

225226
private void succeed(@Nullable ResponseT result) {
@@ -229,7 +230,7 @@ private void succeed(@Nullable ResponseT result) {
229230
}
230231
inFlightFuture = null;
231232
}
232-
closePayload();
233+
closePayload(null);
233234
resultFuture.set(result);
234235
}
235236

@@ -243,19 +244,44 @@ private void fail(Throwable t) {
243244
inFlightFuture = null;
244245
}
245246
inFlight.cancel(true);
247+
ApiException terminal = toTerminalException(t);
246248
progressTracker.onFailed();
247-
closePayload();
248-
resultFuture.setException(t);
249+
closePayload(terminal);
250+
resultFuture.setException(terminal);
249251
}
250252

251-
private void closePayload() {
253+
private ApiException toTerminalException(Throwable t) {
254+
if (t instanceof ApiException && uploadSessionUrl == null) {
255+
return (ApiException) t;
256+
}
257+
String message = firstNonNull(t.getMessage(), t.getClass().getSimpleName());
258+
if (uploadSessionUrl != null) {
259+
message = message + " (upload URL: " + uploadSessionUrl + ")";
260+
}
261+
// Rethrow with URL included in message.
262+
if (t instanceof ApiException) {
263+
ApiException apiException = (ApiException) t;
264+
return ApiExceptionFactory.createException(
265+
message,
266+
apiException.getCause(),
267+
apiException.getStatusCode(),
268+
apiException.isRetryable(),
269+
apiException.getErrorDetails());
270+
}
271+
return ApiExceptionFactory.createException(
272+
message, t, statusCodeOf(StatusCode.Code.UNKNOWN), false);
273+
}
274+
275+
private void closePayload(@Nullable Throwable primaryException) {
252276
if (payload == null) {
253277
return;
254278
}
255279
try {
256280
payload.close();
257-
} catch (IOException ignored) {
258-
// Suppressed during stream cleanup
281+
} catch (Throwable e) {
282+
if (primaryException != null) {
283+
primaryException.addSuppressed(e);
284+
}
259285
}
260286
}
261287

@@ -283,7 +309,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
283309
}
284310
inFlight.cancel(mayInterruptIfRunning);
285311
progressTracker.onFailed();
286-
closePayload();
312+
closePayload(null);
287313
return cancelled;
288314
}
289315

@@ -308,16 +334,17 @@ public ResponseT get(long timeout, TimeUnit unit)
308334
return resultFuture.get(timeout, unit);
309335
}
310336

311-
private static final StatusCode TIMEOUT_STATUS_CODE =
312-
new StatusCode() {
313-
@Override
314-
public StatusCode.Code getCode() {
315-
return StatusCode.Code.DEADLINE_EXCEEDED;
316-
}
337+
private static StatusCode statusCodeOf(StatusCode.Code code) {
338+
return new StatusCode() {
339+
@Override
340+
public StatusCode.Code getCode() {
341+
return code;
342+
}
317343

318-
@Override
319-
public @Nullable Object getTransportCode() {
320-
return null;
321-
}
322-
};
344+
@Override
345+
public @Nullable Object getTransportCode() {
346+
return null;
347+
}
348+
};
349+
}
323350
}

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

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -46,19 +46,17 @@ final class RewindableStreamBuffer {
4646

4747
private final InputStream inputStream;
4848
private final int chunkSize;
49-
private final String uploadUrl;
5049
private final byte[] buffer;
5150

5251
private long bufferBaseOffset;
5352
private int payloadLength;
5453
private boolean isFinal;
5554
private boolean streamExhausted;
5655

57-
RewindableStreamBuffer(InputStream inputStream, int chunkSize, String uploadUrl) {
56+
RewindableStreamBuffer(InputStream inputStream, int chunkSize) {
5857
this.inputStream = checkNotNull(inputStream, "inputStream must not be null");
5958
checkArgument(chunkSize > 0, "chunkSize must be > 0");
6059
this.chunkSize = chunkSize;
61-
this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null");
6260
this.buffer = new byte[chunkSize];
6361
this.bufferBaseOffset = 0L;
6462
this.payloadLength = 0;
@@ -95,17 +93,16 @@ void realignTo(long committedOffset) throws IOException {
9593
if (committedOffset < bufferBaseOffset) {
9694
throw new IllegalStateException(
9795
String.format(
98-
"Server committed offset %d is below buffer base offset %d for upload URL %s; cannot"
99-
+ " rewind stream before buffer base",
100-
committedOffset, bufferBaseOffset, uploadUrl));
96+
"Server committed offset %d is below buffer base offset %d, which the server already"
97+
+ " acknowledged; the upload cannot continue and must be restarted.",
98+
committedOffset, bufferBaseOffset));
10199
}
102100

103101
if (committedOffset > bufferBaseOffset + payloadLength) {
104102
throw new IllegalStateException(
105103
String.format(
106-
"Server committed offset %d is beyond current buffer window [%d, %d] for upload URL"
107-
+ " %s",
108-
committedOffset, bufferBaseOffset, bufferBaseOffset + payloadLength, uploadUrl));
104+
"Server committed offset %d is beyond current buffer window [%d, %d]",
105+
committedOffset, bufferBaseOffset, bufferBaseOffset + payloadLength));
109106
}
110107

111108
int committedWithinBuffer = (int) (committedOffset - bufferBaseOffset);

‎sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java‎

Lines changed: 78 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import com.google.api.gax.rpc.testing.FakeCallContext;
5959
import java.io.ByteArrayInputStream;
6060
import java.io.IOException;
61+
import java.io.InputStream;
6162
import java.nio.charset.StandardCharsets;
6263
import java.time.Duration;
6364
import java.util.List;
@@ -305,7 +306,7 @@ void testUploadCallable_startFailure_failsFuture() {
305306
callable.futureCall("resource-path", payloadOf("data"), null);
306307

307308
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
308-
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
309+
assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
309310
assertThat(exception.getCause()).hasMessageThat().contains("start failed");
310311
verifyNoInteractions(mockChunkCallable);
311312
}
@@ -320,7 +321,7 @@ void testUploadCallable_chunkFailure_failsFuture() {
320321
callable.futureCall("resource-path", payloadOf("data"), null);
321322

322323
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
323-
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
324+
assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
324325
assertThat(exception.getCause()).hasMessageThat().contains("chunk error");
325326
}
326327

@@ -609,7 +610,7 @@ void testRecovery_queryNullOffset_failsFatal() throws Exception {
609610
callable.futureCall("resource-path", payloadOf("hello"), null);
610611

611612
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
612-
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
613+
assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
613614
assertThat(exception.getCause())
614615
.hasMessageThat()
615616
.contains("did not include a committed offset");
@@ -669,7 +670,7 @@ void testRecovery_offsetBelowBufferBase_failsFatal() throws Exception {
669670
callable.futureCall("resource-path", payloadOf("0123456789abcdef"), null);
670671

671672
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
672-
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
673+
assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
673674
assertThat(exception.getCause()).hasMessageThat().contains("below buffer base offset");
674675
}
675676

@@ -746,7 +747,7 @@ void testRecovery_queryMissingStatusHeader_failsFatal() throws Exception {
746747
callable.futureCall("resource-path", payloadOf("hello"), null);
747748

748749
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
749-
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
750+
assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
750751
assertThat(exception.getCause())
751752
.hasMessageThat()
752753
.contains("missing X-Goog-Upload-Status header");
@@ -824,6 +825,7 @@ void testRecovery_serverRejectionWithFinalStatus_failsFatalWithoutRetryOrRecover
824825
assertThat(ex.getCause()).isInstanceOf(InvalidArgumentException.class);
825826
ApiException cause = (ApiException) ex.getCause();
826827
assertThat(cause.getStatusCode().getTransportCode()).isEqualTo(400);
828+
assertThat(cause.getMessage()).contains(sessionUrl);
827829
verify(mockChunkCallable, times(1)).futureCall(any(), any());
828830
verifyNoInteractions(mockQueryCallable);
829831
}
@@ -1259,6 +1261,77 @@ List<ResumableUploadProgress> awaitTerminal() throws InterruptedException {
12591261
}
12601262
}
12611263

1264+
@Test
1265+
void testActionableErrors_startFailure_preservesOriginalException() {
1266+
ApiException startError = createApiException(401, StatusCode.Code.UNAUTHENTICATED);
1267+
when(mockStartCallable.futureCall(any(), any()))
1268+
.thenReturn(ApiFutures.immediateFailedFuture(startError));
1269+
1270+
ResumableUploadFuture<String> future =
1271+
callable.futureCall("resource-path", payloadOf("hello"), null);
1272+
1273+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1274+
assertThat(ex.getCause()).isSameInstanceAs(startError);
1275+
assertThat(future.getUploadSessionUrl()).isNull();
1276+
}
1277+
1278+
@Test
1279+
void testActionableErrors_preservesErrorDetailsAndCauseChain() {
1280+
String sessionUrl = "https://upload.url/chunk-error-details-test";
1281+
stubStartSession(sessionUrl);
1282+
ErrorDetails errorDetails = ErrorDetails.builder().build();
1283+
IOException transportCause = new IOException("403 Forbidden");
1284+
ApiException original =
1285+
ApiExceptionFactory.createException(
1286+
"HTTP 403",
1287+
transportCause,
1288+
new HttpStatusStatusCode(403, StatusCode.Code.PERMISSION_DENIED),
1289+
false,
1290+
errorDetails);
1291+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1292+
.thenReturn(ApiFutures.immediateFailedFuture(original));
1293+
1294+
ResumableUploadFuture<String> future =
1295+
callable.futureCall("resource-path", payloadOf("hello"), null);
1296+
1297+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1298+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1299+
ApiException cause = (ApiException) ex.getCause();
1300+
assertThat(cause.getMessage()).contains(sessionUrl);
1301+
assertThat(cause.getCause()).isSameInstanceAs(transportCause);
1302+
assertThat(cause.getErrorDetails()).isSameInstanceAs(errorDetails);
1303+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1304+
}
1305+
1306+
@Test
1307+
void testUploadCallable_failureOutcome_attachesCloseExceptionViaAddSuppressed() {
1308+
when(mockStartCallable.futureCall(any(), any()))
1309+
.thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("upload failed")));
1310+
1311+
InputStream failingStream =
1312+
new InputStream() {
1313+
@Override
1314+
public int read() {
1315+
return -1;
1316+
}
1317+
1318+
@Override
1319+
public void close() throws IOException {
1320+
throw new IOException("stream close error");
1321+
}
1322+
};
1323+
1324+
ResumableUploadFuture<String> future =
1325+
callable.futureCall("resource-path", () -> failingStream, null);
1326+
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
1327+
assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
1328+
assertThat(exception.getCause().getSuppressed()).asList().hasSize(1);
1329+
assertThat(exception.getCause().getSuppressed()[0]).isInstanceOf(IOException.class);
1330+
assertThat(exception.getCause().getSuppressed()[0])
1331+
.hasMessageThat()
1332+
.contains("stream close error");
1333+
}
1334+
12621335
private static class HttpStatusStatusCode implements StatusCode {
12631336
private final int httpStatus;
12641337
private final StatusCode.Code code;

0 commit comments

Comments
 (0)