Skip to content

Commit 220c398

Browse files
whoweswhowes
committed
feat(gax): surface actionable error messages with upload session URL and stream requirements
Augments terminal failure exceptions with the active upload session URL to aid debugging and session recovery. Clarifies error messages when a server committed offset falls below the buffer base offset.
1 parent 8310b51 commit 220c398

4 files changed

Lines changed: 178 additions & 5 deletions

File tree

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

Lines changed: 38 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -247,14 +247,49 @@ private void finish(@Nullable ResponseT response, @Nullable Throwable error) {
247247
progressTracker.onFinalized(progressTracker.getStatus().getBytesUploaded());
248248
resultFuture.set(response);
249249
} else {
250+
Throwable augmented = augmentWithUrl(error);
250251
if (closeError != null) {
251-
error.addSuppressed(closeError);
252+
augmented.addSuppressed(closeError);
252253
}
253-
progressTracker.onFailed(error, uploadSessionUrl);
254-
resultFuture.setException(error);
254+
progressTracker.onFailed(augmented, uploadSessionUrl);
255+
resultFuture.setException(augmented);
255256
}
256257
}
257258

259+
private Throwable augmentWithUrl(Throwable t) {
260+
String url = uploadSessionUrl;
261+
if (url == null || url.isEmpty()) {
262+
return t;
263+
}
264+
String message = t.getMessage();
265+
if (message != null && message.contains(url)) {
266+
return t;
267+
}
268+
String baseMessage = message != null ? message : t.getClass().getSimpleName();
269+
String augmentedMessage = baseMessage + " (upload URL: " + url + ")";
270+
Throwable augmented = t;
271+
if (t instanceof ApiException) {
272+
ApiException apiException = (ApiException) t;
273+
augmented =
274+
ApiExceptionFactory.createException(
275+
augmentedMessage,
276+
apiException,
277+
apiException.getStatusCode(),
278+
apiException.isRetryable(),
279+
apiException.getErrorDetails());
280+
} else if (t instanceof IllegalStateException) {
281+
augmented = new IllegalStateException(augmentedMessage, t);
282+
} else if (t instanceof IOException) {
283+
augmented = new IOException(augmentedMessage, t);
284+
}
285+
if (augmented != t) {
286+
for (Throwable suppressed : t.getSuppressed()) {
287+
augmented.addSuppressed(suppressed);
288+
}
289+
}
290+
return augmented;
291+
}
292+
258293
private @Nullable IOException closePayload() {
259294
synchronized (lock) {
260295
if (payloadClosed) {

‎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
@@ -101,7 +101,8 @@ void realignTo(long committedOffset) throws IOException {
101101
throw UploadErrors.protocolViolation(
102102
String.format(
103103
"Server committed offset %d is below buffer base offset %d for upload URL %s; cannot"
104-
+ " rewind stream before buffer base",
104+
+ " rewind stream before buffer base. A seekable stream is required to rewind to"
105+
+ " earlier offsets.",
105106
committedOffset, bufferBaseOffset, uploadUrl));
106107
}
107108

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

Lines changed: 137 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,11 @@ void setUp() {
110110
callContext = FakeCallContext.createDefault();
111111
executor = Executors.newScheduledThreadPool(2);
112112
clientContext =
113-
ClientContext.newBuilder().setDefaultCallContext(callContext).setExecutor(executor).build();
113+
ClientContext.newBuilder()
114+
.setDefaultCallContext(callContext)
115+
.setExecutor(executor)
116+
.setEndpoint("https://test.endpoint.com")
117+
.build();
114118
callable = new ResumableUploadCallableImpl<>(mockClient, defaultSettings, clientContext);
115119
}
116120

@@ -1471,6 +1475,138 @@ void testProgressListener_orderingUnderConcurrency_pinsSequentialExecutor() thro
14711475
}
14721476
}
14731477

1478+
@Test
1479+
void testActionableErrors_startFailure_preservesOriginalExceptionWithoutEndpointSuffix() {
1480+
ApiException startError = createApiException(401, StatusCode.Code.UNAUTHENTICATED);
1481+
when(mockStartCallable.futureCall(any(), any()))
1482+
.thenReturn(ApiFutures.immediateFailedFuture(startError));
1483+
1484+
ResumableUploadFuture<String> future =
1485+
callable.futureCall("resource-path", streamOf("hello"), null);
1486+
1487+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1488+
assertThat(ex.getCause()).isSameInstanceAs(startError);
1489+
assertThat(ex.getCause().getMessage()).doesNotContain("endpoint:");
1490+
assertThat(future.getUploadSessionUrl()).isNull();
1491+
}
1492+
1493+
@Test
1494+
void testActionableErrors_chunkFailure_messageContainsUploadSessionUrl() {
1495+
String sessionUrl = "https://upload.url/chunk-error-test";
1496+
stubStartSession(sessionUrl);
1497+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1498+
.thenReturn(
1499+
ApiFutures.immediateFailedFuture(
1500+
createApiException(403, StatusCode.Code.PERMISSION_DENIED)));
1501+
1502+
ResumableUploadFuture<String> future =
1503+
callable.futureCall("resource-path", streamOf("hello"), null);
1504+
1505+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1506+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1507+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1508+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1509+
}
1510+
1511+
@Test
1512+
void testActionableErrors_preservesErrorDetailsCauseChainAndSuppressedExceptions() {
1513+
String sessionUrl = "https://upload.url/chunk-error-details-test";
1514+
stubStartSession(sessionUrl);
1515+
ErrorDetails errorDetails = ErrorDetails.builder().build();
1516+
ApiException original =
1517+
ApiExceptionFactory.createException(
1518+
"HTTP 403",
1519+
null,
1520+
new HttpStatusStatusCode(403, StatusCode.Code.PERMISSION_DENIED),
1521+
false,
1522+
errorDetails);
1523+
IOException suppressed = new IOException("underlying stream error");
1524+
original.addSuppressed(suppressed);
1525+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1526+
.thenReturn(ApiFutures.immediateFailedFuture(original));
1527+
1528+
ResumableUploadFuture<String> future =
1529+
callable.futureCall("resource-path", streamOf("hello"), null);
1530+
1531+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1532+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1533+
ApiException cause = (ApiException) ex.getCause();
1534+
assertThat(cause.getMessage()).contains(sessionUrl);
1535+
assertThat(cause.getCause()).isSameInstanceAs(original);
1536+
assertThat(cause.getErrorDetails()).isSameInstanceAs(errorDetails);
1537+
assertThat(cause.getSuppressed()).asList().contains(suppressed);
1538+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1539+
}
1540+
1541+
@Test
1542+
void testActionableErrors_recoveryFailure_messageContainsUploadSessionUrl() {
1543+
String sessionUrl = "https://upload.url/recovery-error-test";
1544+
stubStartSession(sessionUrl);
1545+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1546+
.thenReturn(
1547+
ApiFutures.immediateFailedFuture(
1548+
createApiException(400, StatusCode.Code.INVALID_ARGUMENT)));
1549+
when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any()))
1550+
.thenReturn(
1551+
ApiFutures.immediateFailedFuture(
1552+
createApiException(403, StatusCode.Code.PERMISSION_DENIED)));
1553+
1554+
ResumableUploadFuture<String> future =
1555+
callable.futureCall("resource-path", streamOf("hello"), null);
1556+
1557+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1558+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1559+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1560+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1561+
}
1562+
1563+
@Test
1564+
void testActionableErrors_globalTimeoutFailure_messageContainsUploadSessionUrl() {
1565+
String sessionUrl = "https://upload.url/timeout-error-test";
1566+
stubStartSession(sessionUrl);
1567+
SettableApiFuture<ChunkUploadResponse<String>> hungChunk = SettableApiFuture.create();
1568+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())).thenReturn(hungChunk);
1569+
1570+
ResumableUploadCallSettings settings =
1571+
defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(50)).build();
1572+
1573+
ResumableUploadFuture<String> future =
1574+
callable.futureCall("resource-path", streamOf("hello"), null, settings);
1575+
1576+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1577+
assertThat(ex.getCause()).isInstanceOf(DeadlineExceededException.class);
1578+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1579+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1580+
}
1581+
1582+
@Test
1583+
void testActionableErrors_rewindFailure_surfacesActionableSeekableStreamMessage() {
1584+
String sessionUrl = "https://upload.url/rewind-error-test";
1585+
stubStartSession(sessionUrl);
1586+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1587+
.thenReturn(
1588+
ApiFutures.immediateFuture(
1589+
ChunkUploadResponse.create(ResumableUploadStatus.ACTIVE, null)))
1590+
.thenReturn(
1591+
ApiFutures.immediateFailedFuture(
1592+
createApiException(400, StatusCode.Code.INVALID_ARGUMENT)));
1593+
1594+
when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any()))
1595+
.thenReturn(
1596+
ApiFutures.immediateFuture(
1597+
createQueryResponse(false, 4L, null, ResumableUploadStatus.ACTIVE)));
1598+
1599+
byte[] data = new byte[16];
1600+
ResumableUploadFuture<String> future =
1601+
callable.futureCall("resource-path", new ByteArrayInputStream(data), null);
1602+
1603+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1604+
assertThat(ex.getCause()).isInstanceOf(FailedPreconditionException.class);
1605+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1606+
assertThat(ex.getCause().getMessage()).contains("seekable stream");
1607+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1608+
}
1609+
14741610
private static class HttpStatusStatusCode implements StatusCode {
14751611
private final int httpStatus;
14761612
private final StatusCode.Code code;

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,7 @@ void testRealignToBelowBaseOffset_throwsFailedPreconditionException_classifiedFa
154154
assertThat(exception.getMessage()).contains("4");
155155
assertThat(exception.getMessage()).contains("8");
156156
assertThat(exception.getMessage()).contains(UPLOAD_URL);
157+
assertThat(exception.getMessage()).contains("seekable stream");
157158

158159
// Must be classified as FATAL by ResumableUploadErrorClassifier
159160
ResumableUploadErrorClassifier.Category category =

0 commit comments

Comments
 (0)