Skip to content

Commit 51ec823

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 469ce0d commit 51ec823

4 files changed

Lines changed: 211 additions & 4 deletions

File tree

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

Lines changed: 41 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -260,9 +260,48 @@ private void fail(Throwable t) {
260260
if (inFlight != null) {
261261
inFlight.cancel(true);
262262
}
263+
Throwable augmented = augmentWithUrl(t);
263264
progressTracker.onFailed();
264-
closePayload();
265-
resultFuture.setException(t);
265+
try {
266+
payload.close();
267+
} catch (Throwable closeException) {
268+
augmented.addSuppressed(closeException);
269+
}
270+
resultFuture.setException(augmented);
271+
}
272+
273+
private Throwable augmentWithUrl(Throwable t) {
274+
String url = uploadSessionUrl;
275+
if (url == null || url.isEmpty()) {
276+
return t;
277+
}
278+
String message = t.getMessage();
279+
if (message != null && message.contains(url)) {
280+
return t;
281+
}
282+
String baseMessage = message != null ? message : t.getClass().getSimpleName();
283+
String augmentedMessage = baseMessage + " (upload URL: " + url + ")";
284+
Throwable augmented = t;
285+
if (t instanceof ApiException) {
286+
ApiException apiException = (ApiException) t;
287+
augmented =
288+
ApiExceptionFactory.createException(
289+
augmentedMessage,
290+
apiException,
291+
apiException.getStatusCode(),
292+
apiException.isRetryable(),
293+
apiException.getErrorDetails());
294+
} else if (t instanceof IllegalStateException) {
295+
augmented = new IllegalStateException(augmentedMessage, t);
296+
} else if (t instanceof IOException) {
297+
augmented = new IOException(augmentedMessage, t);
298+
}
299+
if (augmented != t) {
300+
for (Throwable suppressed : t.getSuppressed()) {
301+
augmented.addSuppressed(suppressed);
302+
}
303+
}
304+
return augmented;
266305
}
267306

268307
private void 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
@@ -95,7 +95,8 @@ void realignTo(long committedOffset) throws IOException {
9595
throw protocolViolation(
9696
String.format(
9797
"Server committed offset %d is below buffer base offset %d for upload URL %s; cannot"
98-
+ " rewind stream before buffer base",
98+
+ " rewind stream before buffer base. A seekable stream is required to rewind to"
99+
+ " earlier offsets.",
99100
committedOffset, bufferBaseOffset, uploadUrl));
100101
}
101102

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

Lines changed: 167 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,11 @@ void setUp() {
108108

109109
defaultSettings = ResumableUploadCallSettings.newBuilder().setChunkSize(8).build();
110110
callContext = FakeCallContext.createDefault();
111-
clientContext = ClientContext.newBuilder().setDefaultCallContext(callContext).build();
111+
clientContext =
112+
ClientContext.newBuilder()
113+
.setDefaultCallContext(callContext)
114+
.setEndpoint("https://test.endpoint.com")
115+
.build();
112116
executor = Executors.newSingleThreadExecutor();
113117
callable = new ResumableUploadCallableImpl<>(mockClient, defaultSettings, clientContext);
114118
}
@@ -1254,6 +1258,168 @@ void testProgressListener_orderingUnderConcurrency_pinsSequentialExecutor() thro
12541258
}
12551259
}
12561260

1261+
@Test
1262+
void testActionableErrors_startFailure_preservesOriginalExceptionWithoutEndpointSuffix() {
1263+
ApiException startError = createApiException(401, StatusCode.Code.UNAUTHENTICATED);
1264+
when(mockStartCallable.futureCall(any(), any()))
1265+
.thenReturn(ApiFutures.immediateFailedFuture(startError));
1266+
1267+
ResumableUploadFuture<String> future =
1268+
callable.futureCall("resource-path", streamOf("hello"), null);
1269+
1270+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1271+
assertThat(ex.getCause()).isSameInstanceAs(startError);
1272+
assertThat(ex.getCause().getMessage()).doesNotContain("endpoint:");
1273+
assertThat(future.getUploadSessionUrl()).isNull();
1274+
}
1275+
1276+
@Test
1277+
void testActionableErrors_chunkFailure_messageContainsUploadSessionUrl() {
1278+
String sessionUrl = "https://upload.url/chunk-error-test";
1279+
stubStartSession(sessionUrl);
1280+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1281+
.thenReturn(
1282+
ApiFutures.immediateFailedFuture(
1283+
createApiException(403, StatusCode.Code.PERMISSION_DENIED)));
1284+
1285+
ResumableUploadFuture<String> future =
1286+
callable.futureCall("resource-path", streamOf("hello"), null);
1287+
1288+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1289+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1290+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1291+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1292+
}
1293+
1294+
@Test
1295+
void testActionableErrors_preservesErrorDetailsCauseChainAndSuppressedExceptions() {
1296+
String sessionUrl = "https://upload.url/chunk-error-details-test";
1297+
stubStartSession(sessionUrl);
1298+
ErrorDetails errorDetails = ErrorDetails.builder().build();
1299+
ApiException original =
1300+
ApiExceptionFactory.createException(
1301+
"HTTP 403",
1302+
null,
1303+
new HttpStatusStatusCode(403, StatusCode.Code.PERMISSION_DENIED),
1304+
false,
1305+
errorDetails);
1306+
IOException suppressed = new IOException("underlying stream error");
1307+
original.addSuppressed(suppressed);
1308+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1309+
.thenReturn(ApiFutures.immediateFailedFuture(original));
1310+
1311+
ResumableUploadFuture<String> future =
1312+
callable.futureCall("resource-path", streamOf("hello"), null);
1313+
1314+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1315+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1316+
ApiException cause = (ApiException) ex.getCause();
1317+
assertThat(cause.getMessage()).contains(sessionUrl);
1318+
assertThat(cause.getCause()).isSameInstanceAs(original);
1319+
assertThat(cause.getErrorDetails()).isSameInstanceAs(errorDetails);
1320+
assertThat(cause.getSuppressed()).asList().contains(suppressed);
1321+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1322+
}
1323+
1324+
@Test
1325+
void testActionableErrors_recoveryFailure_messageContainsUploadSessionUrl() {
1326+
String sessionUrl = "https://upload.url/recovery-error-test";
1327+
stubStartSession(sessionUrl);
1328+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1329+
.thenReturn(
1330+
ApiFutures.immediateFailedFuture(
1331+
createApiException(400, StatusCode.Code.INVALID_ARGUMENT)));
1332+
when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any()))
1333+
.thenReturn(
1334+
ApiFutures.immediateFailedFuture(
1335+
createApiException(403, StatusCode.Code.PERMISSION_DENIED)));
1336+
1337+
ResumableUploadFuture<String> future =
1338+
callable.futureCall("resource-path", streamOf("hello"), null);
1339+
1340+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1341+
assertThat(ex.getCause()).isInstanceOf(ApiException.class);
1342+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1343+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1344+
}
1345+
1346+
@Test
1347+
void testActionableErrors_globalTimeoutFailure_messageContainsUploadSessionUrl() {
1348+
String sessionUrl = "https://upload.url/timeout-error-test";
1349+
stubStartSession(sessionUrl);
1350+
SettableApiFuture<ChunkUploadResponse<String>> hungChunk = SettableApiFuture.create();
1351+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())).thenReturn(hungChunk);
1352+
1353+
ResumableUploadCallSettings settings =
1354+
defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(50)).build();
1355+
1356+
ResumableUploadFuture<String> future =
1357+
callable.futureCall("resource-path", streamOf("hello"), null, settings);
1358+
1359+
ExecutionException ex =
1360+
assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS));
1361+
assertThat(ex.getCause()).isInstanceOf(DeadlineExceededException.class);
1362+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1363+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1364+
}
1365+
1366+
@Test
1367+
void testActionableErrors_rewindFailure_surfacesActionableSeekableStreamMessage() {
1368+
String sessionUrl = "https://upload.url/rewind-error-test";
1369+
stubStartSession(sessionUrl);
1370+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
1371+
.thenReturn(
1372+
ApiFutures.immediateFuture(
1373+
ChunkUploadResponse.create(ResumableUploadStatus.ACTIVE, null)))
1374+
.thenReturn(
1375+
ApiFutures.immediateFailedFuture(
1376+
createApiException(400, StatusCode.Code.INVALID_ARGUMENT)));
1377+
1378+
when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any()))
1379+
.thenReturn(
1380+
ApiFutures.immediateFuture(
1381+
createQueryResponse(4L, null, ResumableUploadStatus.ACTIVE)));
1382+
1383+
byte[] data = new byte[16];
1384+
ResumableUploadFuture<String> future =
1385+
callable.futureCall("resource-path", new ByteArrayInputStream(data), null);
1386+
1387+
ExecutionException ex = assertThrows(ExecutionException.class, future::get);
1388+
assertThat(ex.getCause()).isInstanceOf(FailedPreconditionException.class);
1389+
assertThat(ex.getCause().getMessage()).contains(sessionUrl);
1390+
assertThat(ex.getCause().getMessage()).contains("seekable stream");
1391+
assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
1392+
}
1393+
1394+
@Test
1395+
void testUploadCallable_failureOutcome_attachesCloseExceptionViaAddSuppressed() {
1396+
when(mockStartCallable.futureCall(any(), any()))
1397+
.thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("upload failed")));
1398+
1399+
InputStream failingStream =
1400+
new InputStream() {
1401+
@Override
1402+
public int read() {
1403+
return -1;
1404+
}
1405+
1406+
@Override
1407+
public void close() throws IOException {
1408+
throw new IOException("stream close error");
1409+
}
1410+
};
1411+
1412+
ResumableUploadFuture<String> future =
1413+
callable.futureCall("resource-path", failingStream, null);
1414+
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
1415+
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
1416+
assertThat(exception.getCause().getSuppressed()).asList().hasSize(1);
1417+
assertThat(exception.getCause().getSuppressed()[0]).isInstanceOf(IOException.class);
1418+
assertThat(exception.getCause().getSuppressed()[0])
1419+
.hasMessageThat()
1420+
.contains("stream close error");
1421+
}
1422+
12571423
private static class HttpStatusStatusCode implements StatusCode {
12581424
private final int httpStatus;
12591425
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
@@ -150,6 +150,7 @@ void testRealignToBelowBaseOffset_throwsFailedPreconditionException() throws IOE
150150
assertThat(exception.getMessage()).contains("4");
151151
assertThat(exception.getMessage()).contains("8");
152152
assertThat(exception.getMessage()).contains(UPLOAD_URL);
153+
assertThat(exception.getMessage()).contains("seekable stream");
153154
}
154155

155156
@Test

0 commit comments

Comments
 (0)