Skip to content

Commit cff442f

Browse files
whoweswhowes
committed
feat(gax): surface actionable error messages with upload session URL and stream requirements
Improve error messages across resumable upload failure paths to provide actionable context for debugging and recovery. - In ResumableUploadFutureImpl, augment outgoing exception messages on terminal failures with the active upload session URL (or the endpoint if session initiation failed), ensuring callers have the session URL required for diagnostic queries and manual resume. - In RewindableStreamBuffer, clarify the UploadProtocolViolationException message when a server committed offset falls below the buffer base offset by explicitly explaining that a seekable stream is required to rewind to earlier offsets. - In ResumableUploadChunkCoordinator, include the upload session URL when reporting an incomplete upload status after transmitting the final chunk. - Add comprehensive unit tests in ResumableUploadCallableImplTest and RewindableStreamBufferTest verifying that the upload URL or endpoint appears in error messages across start failures, chunk transfer failures, recovery query failures, global timeout expirations, and rewind-below-base violations.
1 parent ca9311c commit cff442f

8 files changed

Lines changed: 505 additions & 12 deletions

File tree

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

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -105,13 +105,14 @@ void setRetryingFuture(RetryingFuture<ChunkUploadResponse<ResponseT>> retryingFu
105105

106106
private boolean needsRecovery() {
107107
if (lastFailure != null) {
108-
UploadErrorCategory category = UploadErrorClassifier.classify(lastFailure, currentCommand);
109-
return category == UploadErrorCategory.RECOVERABLE;
108+
UploadErrorClassifier.Category category =
109+
UploadErrorClassifier.classify(lastFailure, currentCommand);
110+
return category == UploadErrorClassifier.Category.RECOVERABLE;
110111
}
111112
if (lastResponse != null && lastResponse.getUploadStatus() == null) {
112-
UploadErrorCategory category =
113+
UploadErrorClassifier.Category category =
113114
UploadErrorClassifier.classifyMissingStatusHeader(currentCommand);
114-
return category == UploadErrorCategory.RECOVERABLE;
115+
return category == UploadErrorClassifier.Category.RECOVERABLE;
115116
}
116117
return false;
117118
}

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

Lines changed: 40 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,7 @@ public StatusCode.Code getCode() {
198198
return null;
199199
}
200200
};
201+
201202
private void onTimeout() {
202203
synchronized (lock) {
203204
if (done) {
@@ -282,12 +283,48 @@ private void finish(@Nullable ResponseT response, @Nullable Throwable error) {
282283
progressTracker.onFinalized(totalBytes);
283284
result.set(response);
284285
} else {
285-
progressTracker.onFailed(error, uploadSessionUrl);
286+
Throwable augmented = augmentWithUrl(error);
286287
if (closeError != null) {
287-
error.addSuppressed(closeError);
288+
augmented.addSuppressed(closeError);
289+
}
290+
progressTracker.onFailed(augmented, uploadSessionUrl);
291+
result.setException(augmented);
292+
}
293+
}
294+
295+
private Throwable augmentWithUrl(Throwable t) {
296+
String url = uploadSessionUrl != null ? uploadSessionUrl : clientContext.getEndpoint();
297+
if (url == null || url.isEmpty()) {
298+
return t;
299+
}
300+
String message = t.getMessage();
301+
if (message != null && message.contains(url)) {
302+
return t;
303+
}
304+
String label = uploadSessionUrl != null ? "upload URL: " : "endpoint: ";
305+
String augmentedMessage =
306+
(message != null ? message : t.getClass().getSimpleName()) + " (" + label + url + ")";
307+
Throwable augmented = t;
308+
if (t instanceof ApiException) {
309+
ApiException apiException = (ApiException) t;
310+
augmented =
311+
ApiExceptionFactory.createException(
312+
augmentedMessage,
313+
apiException.getCause(),
314+
apiException.getStatusCode(),
315+
apiException.isRetryable(),
316+
apiException.getErrorDetails());
317+
} else if (t instanceof IllegalStateException) {
318+
augmented = new IllegalStateException(augmentedMessage, t.getCause());
319+
} else if (t instanceof IOException) {
320+
augmented = new IOException(augmentedMessage, t.getCause());
321+
}
322+
if (augmented != t) {
323+
for (Throwable suppressed : t.getSuppressed()) {
324+
augmented.addSuppressed(suppressed);
288325
}
289-
result.setException(error);
290326
}
327+
return augmented;
291328
}
292329

293330
private @Nullable IOException closePayload() {

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,8 @@ void realignTo(long committedOffset) throws IOException {
102102
throw protocolViolation(
103103
String.format(
104104
"Server committed offset %d is below buffer base offset %d for upload URL %s; cannot"
105-
+ " rewind stream before buffer base",
105+
+ " rewind stream before buffer base. A seekable stream is required to rewind to"
106+
+ " earlier offsets.",
106107
committedOffset, bufferBaseOffset, uploadUrl));
107108
}
108109

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -237,7 +237,10 @@ private static void dispatchSafely(
237237
try {
238238
listener.onProgress(status);
239239
} catch (Throwable t) {
240-
LOG.log(Level.WARNING, "Exception thrown by ResumableUploadProgressListener", t);
240+
LOG.log(
241+
Level.WARNING,
242+
"Exception thrown by ResumableUploadProgressListener: " + listener.getClass().getName(),
243+
t);
241244
}
242245
}
243246
}

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ public boolean shouldRetry(
6464
if (previousThrowable instanceof CancellationException) {
6565
return false;
6666
}
67-
UploadErrorCategory category = null;
67+
UploadErrorClassifier.Category category = null;
6868
if (previousThrowable != null) {
6969
category = UploadErrorClassifier.classify(previousThrowable, command);
7070
} else if (previousResponse instanceof ChunkUploadResponse) {
@@ -78,7 +78,8 @@ public boolean shouldRetry(
7878
category = UploadErrorClassifier.classifyMissingStatusHeader(command);
7979
}
8080
}
81-
return category == UploadErrorCategory.TRANSIENT || category == UploadErrorCategory.RECOVERABLE;
81+
return category == UploadErrorClassifier.Category.TRANSIENT
82+
|| category == UploadErrorClassifier.Category.RECOVERABLE;
8283
}
8384

8485
@Override

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

Lines changed: 133 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,11 @@ void setUp() {
107107
callContext = FakeCallContext.createDefault();
108108
executor = Executors.newScheduledThreadPool(2);
109109
clientContext =
110-
ClientContext.newBuilder().setDefaultCallContext(callContext).setExecutor(executor).build();
110+
ClientContext.newBuilder()
111+
.setDefaultCallContext(callContext)
112+
.setExecutor(executor)
113+
.setEndpoint("https://test.endpoint.com")
114+
.build();
111115
callable = new ResumableUploadCallableImpl<>(mockClient, defaultSettings, clientContext);
112116
}
113117

@@ -1409,6 +1413,134 @@ void testProgressListener_orderingUnderConcurrency_pinsSequentialExecutor() thro
14091413
}
14101414
}
14111415

1416+
@Test
1417+
void testActionableErrors_startFailureAfterRetries_messageContainsEndpointUrl() {
1418+
when(mockStartCallable.futureCall(any(), any()))
1419+
.thenReturn(
1420+
ApiFutures.immediateFailedFuture(
1421+
createApiException(401, StatusCode.Code.UNAUTHENTICATED)));
1422+
1423+
ResumableUploadFuture<String> future =
1424+
callable.futureCall("resource-path", streamOf("hello"), null);
1425+
1426+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1427+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1428+
assertThat(ex.getCause().getMessage()).contains("https://test.endpoint.com");
1429+
assertThat(future.getUploadSessionUrl()).isNull();
1430+
}
1431+
1432+
@Test
1433+
void testActionableErrors_chunkFailureAfterRetries_messageContainsUploadSessionUrl() {
1434+
String sessionUrl = "https://upload.url/chunk-error-test";
1435+
stubStartSession(sessionUrl);
1436+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1437+
.thenReturn(
1438+
ApiFutures.immediateFailedFuture(
1439+
createApiException(403, StatusCode.Code.PERMISSION_DENIED)));
1440+
1441+
ResumableUploadFuture<String> future =
1442+
callable.futureCall("resource-path", streamOf("hello"), null);
1443+
1444+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1445+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1446+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1447+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1448+
}
1449+
1450+
@Test
1451+
void testActionableErrors_preservesErrorDetailsAndSuppressedExceptions() {
1452+
String sessionUrl = "https://upload.url/chunk-error-details-test";
1453+
stubStartSession(sessionUrl);
1454+
ErrorDetails errorDetails = ErrorDetails.builder().build();
1455+
ApiException original =
1456+
ApiExceptionFactory.createException(
1457+
"HTTP 403",
1458+
null,
1459+
new HttpStatusStatusCode(403, StatusCode.Code.PERMISSION_DENIED),
1460+
false,
1461+
errorDetails);
1462+
IOException suppressed = new IOException("underlying stream error");
1463+
original.addSuppressed(suppressed);
1464+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1465+
.thenReturn(ApiFutures.immediateFailedFuture(original));
1466+
1467+
ResumableUploadFuture<String> future =
1468+
callable.futureCall("resource-path", streamOf("hello"), null);
1469+
1470+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1471+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1472+
ApiException cause = (ApiException) ex.getCause();
1473+
assertThat(cause.getMessage()).contains(sessionUrl);
1474+
assertThat(cause.getErrorDetails()).isSameInstanceAs(errorDetails);
1475+
assertThat(cause.getSuppressed()).asList().contains(suppressed);
1476+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1477+
}
1478+
1479+
@Test
1480+
void testActionableErrors_recoveryFailure_messageContainsUploadSessionUrl() {
1481+
String sessionUrl = "https://upload.url/recovery-error-test";
1482+
stubStartSession(sessionUrl);
1483+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1484+
.thenReturn(
1485+
ApiFutures.immediateFailedFuture(
1486+
createApiException(400, StatusCode.Code.INVALID_ARGUMENT)));
1487+
when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any()))
1488+
.thenReturn(
1489+
ApiFutures.immediateFailedFuture(
1490+
createApiException(403, StatusCode.Code.PERMISSION_DENIED)));
1491+
1492+
ResumableUploadFuture<String> future =
1493+
callable.futureCall("resource-path", streamOf("hello"), null);
1494+
1495+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1496+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1497+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1498+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1499+
}
1500+
1501+
@Test
1502+
void testActionableErrors_globalTimeoutFailure_messageContainsUploadSessionUrl() {
1503+
String sessionUrl = "https://upload.url/timeout-error-test";
1504+
stubStartSession(sessionUrl);
1505+
SettableApiFuture<ChunkUploadResponse<String>> hungChunk = SettableApiFuture.create();
1506+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())).thenReturn(hungChunk);
1507+
1508+
ResumableUploadCallSettings settings =
1509+
defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(50)).build();
1510+
1511+
ResumableUploadFuture<String> future =
1512+
callable.futureCall("resource-path", streamOf("hello"), null, settings);
1513+
1514+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1515+
assertThat(ex.getCause()).isInstanceOf(DeadlineExceededException.class);
1516+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1517+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1518+
}
1519+
1520+
@Test
1521+
void testActionableErrors_rewindFailure_surfacesActionableSeekableStreamMessage() {
1522+
String sessionUrl = "https://upload.url/rewind-error-test";
1523+
stubStartSession(sessionUrl);
1524+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1525+
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(false, null)))
1526+
.thenReturn(
1527+
ApiFutures.immediateFailedFuture(
1528+
createApiException(400, StatusCode.Code.INVALID_ARGUMENT)));
1529+
1530+
when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any()))
1531+
.thenReturn(ApiFutures.immediateFuture(createQueryResponse(false, 4L, null, "active")));
1532+
1533+
byte[] data = new byte[16];
1534+
ResumableUploadFuture<String> future =
1535+
callable.futureCall("resource-path", new ByteArrayInputStream(data), null);
1536+
1537+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1538+
assertThat(ex.getCause()).isInstanceOf(FailedPreconditionException.class);
1539+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1540+
assertThat(ex.getCause().getMessage()).contains("seekable stream");
1541+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1542+
}
1543+
14121544
private static class HttpStatusStatusCode implements StatusCode {
14131545
private final int httpStatus;
14141546
private final StatusCode.Code code;

0 commit comments

Comments
 (0)