This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 4731dbc5a38 Fix RequestResponseIO parseAndThrow to preserve retryable
exception types (#37342)
4731dbc5a38 is described below
commit 4731dbc5a38b8d211ed1ac094749f0cb8425edb1
Author: ZIHAN DAI <[email protected]>
AuthorDate: Tue Aug 11 10:51:06 2026 +1000
Fix RequestResponseIO parseAndThrow to preserve retryable exception types
(#37342)
---
.../org/apache/beam/io/requestresponse/Call.java | 11 ++---
.../apache/beam/io/requestresponse/CallTest.java | 51 +++++++++++++++++++++-
2 files changed, 53 insertions(+), 9 deletions(-)
diff --git
a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java
b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java
index 616a178d1c3..5b9fa496842 100644
---
a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java
+++
b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java
@@ -596,13 +596,10 @@ class Call<RequestT, ResponseT> extends
PTransform<PCollection<RequestT>, Result
private static <T> void parseAndThrow(Future<T> future, ExecutionException e)
throws UserCodeExecutionException {
future.cancel(true);
- if (e.getCause() == null) {
- throw new UserCodeExecutionException(e);
+ Throwable cause = e.getCause();
+ if (cause instanceof UserCodeExecutionException) {
+ throw (UserCodeExecutionException) cause;
}
- Throwable cause = checkStateNotNull(e.getCause());
- if (cause instanceof UserCodeQuotaException) {
- throw new UserCodeQuotaException(cause);
- }
- throw new UserCodeExecutionException(cause);
+ throw new UserCodeExecutionException(cause == null ? e : cause);
}
}
diff --git
a/sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java
b/sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java
index 0764ab8db40..5fb20bf38b8 100644
---
a/sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java
+++
b/sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java
@@ -104,6 +104,20 @@ public class CallTest {
pipeline.run();
}
+ @Test
+ public void
givenCallerThrowsNonUserCodeException_emitsWrappedUserCodeExecutionException() {
+ Result<Response> result =
+ pipeline
+ .apply(Create.of(new Request("a")))
+ .apply(Call.of(new CallerThrowsRuntimeException(),
NON_DETERMINISTIC_RESPONSE_CODER));
+
+ PCollection<ApiIOError> failures = result.getFailures();
+ PAssert.thatSingleton(countStackTracesOf(failures,
UserCodeExecutionException.class))
+ .isEqualTo(1L);
+
+ pipeline.run();
+ }
+
@Test
public void givenCallerThrowsQuotaException_emitsIntoFailurePCollection() {
Result<Response> result =
@@ -142,7 +156,7 @@ public class CallTest {
}
@Test
- public void givenCallerThrowsTimeoutException_emitsFailurePCollection() {
+ public void givenCallerThrowsTimeoutException_thenPreservesExceptionType() {
Result<Response> result =
pipeline
.apply(Create.of(new Request("a")))
@@ -150,7 +164,7 @@ public class CallTest {
PCollection<ApiIOError> failures = result.getFailures();
PAssert.thatSingleton(countStackTracesOf(failures,
UserCodeExecutionException.class))
- .isEqualTo(1L);
+ .isEqualTo(0L);
PAssert.thatSingleton(countStackTracesOf(failures,
UserCodeQuotaException.class)).isEqualTo(0L);
PAssert.thatSingleton(countStackTracesOf(failures,
UserCodeTimeoutException.class))
.isEqualTo(1L);
@@ -158,6 +172,23 @@ public class CallTest {
pipeline.run();
}
+ @Test
+ public void
givenCallerThrowsRemoteSystemException_thenPreservesExceptionType() {
+ Result<Response> result =
+ pipeline
+ .apply(Create.of(new Request("a")))
+ .apply(
+ Call.of(new CallerThrowsRemoteSystemException(),
NON_DETERMINISTIC_RESPONSE_CODER));
+
+ PCollection<ApiIOError> failures = result.getFailures();
+ PAssert.thatSingleton(countStackTracesOf(failures,
UserCodeRemoteSystemException.class))
+ .isEqualTo(1L);
+ PAssert.thatSingleton(countStackTracesOf(failures,
UserCodeExecutionException.class))
+ .isEqualTo(0L);
+
+ pipeline.run();
+ }
+
@Test
public void givenSetupThrowsUserCodeExecutionException_throwsError() {
pipeline
@@ -375,6 +406,14 @@ public class CallTest {
}
}
+ private static class CallerThrowsRuntimeException implements Caller<Request,
Response> {
+
+ @Override
+ public Response call(Request request) {
+ throw new RuntimeException("unexpected error");
+ }
+ }
+
private static class CallerThrowsTimeout implements Caller<Request,
Response> {
@Override
@@ -383,6 +422,14 @@ public class CallTest {
}
}
+ private static class CallerThrowsRemoteSystemException implements
Caller<Request, Response> {
+
+ @Override
+ public Response call(Request request) throws UserCodeExecutionException {
+ throw new UserCodeRemoteSystemException("");
+ }
+ }
+
private static class CallerInvokesQuotaException implements Caller<Request,
Response> {
@Override
