Simplify parseAndThrow: check e.getCause() instead of traversing causal chain

Future.get() wraps with exactly one ExecutionException, so inspecting
e.getCause() is sufficient. If the cause is a UserCodeExecutionException
(or subclass), rethrow it directly — this preserves all retryable types
and is automatically future-proof for new subclasses.

Remove deep-nesting tests that tested the now-removed causal chain
traversal, as those scenarios cannot occur in this code path.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: PDGGK

sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java

@@ -46,7 +46,6 @@
 import org.apache.beam.sdk.values.PCollectionTuple;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.sdk.values.TupleTagList;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.joda.time.Duration;
 
@@ -614,33 +613,10 @@ private void executeAsync(Callable<Void> callable) throws UserCodeExecutionExcep
   private static <T> void parseAndThrow(Future<T> future, ExecutionException e)
       throws UserCodeExecutionException {
     future.cancel(true);
-
-    try {
-      UserCodeExecutionException genericException = null;
-      for (Throwable throwable : Throwables.getCausalChain(e)) {
-        if (throwable instanceof UserCodeQuotaException) {
-          throw (UserCodeQuotaException) throwable;
-        } else if (throwable instanceof UserCodeTimeoutException) {
-          throw (UserCodeTimeoutException) throwable;
-        } else if (throwable instanceof UserCodeRemoteSystemException) {
-          throw (UserCodeRemoteSystemException) throwable;
-        } else if (genericException == null && throwable instanceof UserCodeExecutionException) {
-          genericException = (UserCodeExecutionException) throwable;
-        }
-      }
-      if (genericException != null) {
-        throw genericException;
-      }
-    } catch (IllegalArgumentException iae) {
-      // Circular reference detected in causal chain
-      throw new UserCodeExecutionException(
-          "circular reference detected in exception causal chain", e);
-    }
-
     Throwable cause = e.getCause();
-    if (cause == null) {
-      throw new UserCodeExecutionException(e);
+    if (cause instanceof UserCodeExecutionException) {
+      throw (UserCodeExecutionException) cause;
     }
-    throw new UserCodeExecutionException(cause);
+    throw new UserCodeExecutionException(cause == null ? e : cause);
   }
 }

sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java

@@ -118,20 +118,6 @@ public void givenCallerThrowsNonUserCodeException_emitsWrappedUserCodeExecutionE
     pipeline.run();
   }
 
-  @Test
-  public void givenCallerThrowsCircularCausalChain_emitsUserCodeExecutionException() {
-    Result<Response> result =
-        pipeline
-            .apply(Create.of(new Request("a")))
-            .apply(Call.of(new CallerThrowsCircularCause(), 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 =
@@ -203,136 +189,6 @@ public void givenCallerThrowsRemoteSystemException_thenPreservesExceptionType()
     pipeline.run();
   }
 
-  @Test
-  public void givenNestedExecutionException_thenPreservesExceptionType() {
-    Result<Response> result =
-        pipeline
-            .apply(Create.of(new Request("a")))
-            .apply(
-                Call.of(
-                    new CallerThrowsNestedExecutionException(), NON_DETERMINISTIC_RESPONSE_CODER));
-
-    PCollection<ApiIOError> failures = result.getFailures();
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
-        .isEqualTo(1L);
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeTimeoutException.class))
-        .isEqualTo(0L);
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeRemoteSystemException.class))
-        .isEqualTo(0L);
-
-    pipeline.run();
-  }
-
-  @Test
-  public void givenCallerThrowsGenericWrappingTimeout_thenPreservesExceptionType() {
-    Result<Response> result =
-        pipeline
-            .apply(Create.of(new Request("a")))
-            .apply(
-                Call.of(
-                    new CallerThrowsGenericWrappingTimeout(), NON_DETERMINISTIC_RESPONSE_CODER));
-
-    PCollection<ApiIOError> failures = result.getFailures();
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeTimeoutException.class))
-        .isEqualTo(1L);
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
-        .isEqualTo(0L);
-
-    pipeline.run();
-  }
-
-  @Test
-  public void givenCallerThrowsGenericWrappingQuota_thenPreservesExceptionType() {
-    Result<Response> result =
-        pipeline
-            .apply(Create.of(new Request("a")))
-            .apply(
-                Call.of(new CallerThrowsGenericWrappingQuota(), NON_DETERMINISTIC_RESPONSE_CODER));
-
-    PCollection<ApiIOError> failures = result.getFailures();
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeQuotaException.class)).isEqualTo(1L);
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
-        .isEqualTo(0L);
-
-    pipeline.run();
-  }
-
-  @Test
-  public void givenCallerThrowsGenericWrappingRemoteSystem_thenPreservesExceptionType() {
-    Result<Response> result =
-        pipeline
-            .apply(Create.of(new Request("a")))
-            .apply(
-                Call.of(
-                    new CallerThrowsGenericWrappingRemoteSystem(),
-                    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
-      givenCallerThrowsUncheckedExecutionExceptionWrappingTimeout_thenPreservesExceptionType() {
-    Result<Response> result =
-        pipeline
-            .apply(Create.of(new Request("a")))
-            .apply(
-                Call.of(
-                    new CallerThrowsUncheckedExecutionExceptionWrappingTimeout(),
-                    NON_DETERMINISTIC_RESPONSE_CODER));
-
-    PCollection<ApiIOError> failures = result.getFailures();
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeTimeoutException.class))
-        .isEqualTo(1L);
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
-        .isEqualTo(0L);
-
-    pipeline.run();
-  }
-
-  @Test
-  public void
-      givenCallerThrowsUncheckedExecutionExceptionWrappingRemoteSystem_thenPreservesExceptionType() {
-    Result<Response> result =
-        pipeline
-            .apply(Create.of(new Request("a")))
-            .apply(
-                Call.of(
-                    new CallerThrowsUncheckedExecutionExceptionWrappingRemoteSystem(),
-                    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 givenCallerThrowsTripleNestedTimeout_thenPreservesExceptionType() {
-    Result<Response> result =
-        pipeline
-            .apply(Create.of(new Request("a")))
-            .apply(
-                Call.of(new CallerThrowsTripleNestedTimeout(), NON_DETERMINISTIC_RESPONSE_CODER));
-
-    PCollection<ApiIOError> failures = result.getFailures();
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeTimeoutException.class))
-        .isEqualTo(1L);
-    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
-        .isEqualTo(0L);
-
-    pipeline.run();
-  }
-
   @Test
   public void givenSetupThrowsUserCodeExecutionException_throwsError() {
     pipeline
@@ -559,17 +415,6 @@ public Response call(Request request) {
     }
   }
 
-  private static class CallerThrowsCircularCause implements Caller<Request, Response> {
-
-    @Override
-    public Response call(Request request) {
-      Exception a = new Exception("a");
-      Exception b = new Exception("b", a);
-      a.initCause(b); // a -> b -> a (circular reference)
-      throw new RuntimeException("boom", a);
-    }
-  }
-
   private static class CallerThrowsTimeout implements Caller<Request, Response> {
 
     @Override
@@ -586,66 +431,6 @@ public Response call(Request request) throws UserCodeExecutionException {
     }
   }
 
-  private static class CallerThrowsNestedExecutionException implements Caller<Request, Response> {
-
-    @Override
-    public Response call(Request request) throws UserCodeExecutionException {
-      throw new UncheckedExecutionException(new UserCodeExecutionException("nested"));
-    }
-  }
-
-  private static class CallerThrowsGenericWrappingTimeout implements Caller<Request, Response> {
-
-    @Override
-    public Response call(Request request) throws UserCodeExecutionException {
-      throw new UserCodeExecutionException("generic", new UserCodeTimeoutException("timeout"));
-    }
-  }
-
-  private static class CallerThrowsGenericWrappingQuota implements Caller<Request, Response> {
-
-    @Override
-    public Response call(Request request) throws UserCodeExecutionException {
-      throw new UserCodeExecutionException("generic", new UserCodeQuotaException("quota"));
-    }
-  }
-
-  private static class CallerThrowsGenericWrappingRemoteSystem
-      implements Caller<Request, Response> {
-
-    @Override
-    public Response call(Request request) throws UserCodeExecutionException {
-      throw new UserCodeExecutionException("generic", new UserCodeRemoteSystemException("remote"));
-    }
-  }
-
-  private static class CallerThrowsUncheckedExecutionExceptionWrappingTimeout
-      implements Caller<Request, Response> {
-
-    @Override
-    public Response call(Request request) throws UserCodeExecutionException {
-      throw new UncheckedExecutionException(new UserCodeTimeoutException("timeout"));
-    }
-  }
-
-  private static class CallerThrowsUncheckedExecutionExceptionWrappingRemoteSystem
-      implements Caller<Request, Response> {
-
-    @Override
-    public Response call(Request request) throws UserCodeExecutionException {
-      throw new UncheckedExecutionException(new UserCodeRemoteSystemException("remote"));
-    }
-  }
-
-  private static class CallerThrowsTripleNestedTimeout implements Caller<Request, Response> {
-
-    @Override
-    public Response call(Request request) throws UserCodeExecutionException {
-      throw new UncheckedExecutionException(
-          new RuntimeException(new UserCodeTimeoutException("deep timeout")));
-    }
-  }
-
   private static class CallerInvokesQuotaException implements Caller<Request, Response> {
 
     @Override