Repository: quarkusio/quarkus
Issue: Async Events behave inconsistently on Exceptions

### Describe the bug

If an event is fired (and observed) asynchronously, the returned CompletionStage behaves differently in the case of exceptions depending on the number of observers that threw exceptions. This violates the [CDI Spec, 10.5.1](https://jakarta.ee/specifications/cdi/3.0/jakarta-cdi-spec-3.0.html#async_exception).

### Expected behavior

Both test methods should be green.

### Actual behavior

The test case  `testSingleFailure` fails, because the `CompletionStage` contains the `IllegalArgumentException` directly instead of a `CompletionException` containing the `IllegalArgumentException` as the spec requires.

```
java.lang.AssertionError: 
Expecting actual throwable to be an instance of:
  java.util.concurrent.CompletionException
but was:
  java.lang.IllegalArgumentException
	at com.example.EventTest$Observer1.failOn42And47(EventDecoratorTest.java:71)
	at com.example.EventTest_Observer1_Observer_failOn42And47_df18bd4f87120ec5a98ebc09b259d85482e10edc.notify(Unknown Source)
	at io.quarkus.arc.impl.EventImpl$Notifier.notifyObservers(EventImpl.java:320)
	...(10 remaining lines not displayed - this can be changed with Assertions.setMaxStackTraceElementsDisplayed)

	at com.example.EventTest.testSingleFailure(EventDecoratorTest.java:43)
```



### How to Reproduce?

```
// imports

@QuarkusTest
class EventTest {

    @Inject
    Observer1 observer1;

    @Inject
    Observer2 observer2;

    @Inject
    Event<Integer> event;

    @Test
    void testSingleFailure() throws InterruptedException, TimeoutException {
        final CompletionStage<Integer> stage = event.fireAsync(42);

        try {
            stage.toCompletableFuture().get(30, TimeUnit.SECONDS);
            failBecauseExceptionWasNotThrown(ExecutionException.class);
        } catch (ExecutionException ex) {
            assertThat(ex.getCause()).isInstanceOf(CompletionException.class)
                    .extracting(Throwable::getSuppressed, as(ARRAY))
                    .hasSize(1)
                    .hasAtLeastOneElementOfType(IllegalArgumentException.class);
        }
    }

    @Test
    void testTwoFailures() throws InterruptedException, TimeoutException {
        final CompletionStage<Integer> stage = event.fireAsync(47);

        try {
            stage.toCompletableFuture().get(30, TimeUnit.SECONDS);
            failBecauseExceptionWasNotThrown(ExecutionException.class);
        } catch (ExecutionException ex) {
            assertThat(ex.getCause()).isInstanceOf(CompletionException.class)
                    .extracting(Throwable::getSuppressed, as(ARRAY))
                    .hasSize(2)
                    .hasAtLeastOneElementOfType(IllegalArgumentException.class)
                    .hasAtLeastOneElementOfType(IllegalStateException.class);
        }
    }

    @ApplicationScoped
    static class Observer1 {

        void failOn42And47(@ObservesAsync int value) {
            if (value == 42 || value == 47) {
                throw new IllegalArgumentException();
            }
        }
    }

    @ApplicationScoped
    static class Observer2 {

        void failOn47(@ObservesAsync int value) {
            if (value == 47) {
                throw new IllegalStateException();
            }
        }
    }
}
```

### Output of `uname -a` or `ver`

_No response_

### Output of `java -version`

11.0.11, vendor: AdoptOpenJDK, runtime: C:\\Program Files\\AdoptOpenJDK\\jdk-11.0.11.9-hotspot

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

2.7.5.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

Apache Maven 3.8.1 (05c21c65bdfed0f71a2f2ada8b84da59348c4c5d)

### Additional information

_No response_

Changed files:
independent-projects/arc/tests/src/test/java/io/quarkus/arc/test/observers/async/AsyncObserverExceptionTest.java
independent-projects/arc/runtime/src/main/java/io/quarkus/arc/impl/EventImpl.java

diff --git a/independent-projects/arc/runtime/src/main/java/io/quarkus/arc/impl/EventImpl.java b/independent-projects/arc/runtime/src/main/java/io/quarkus/arc/impl/EventImpl.java
index 9168cd4469b..efe44c3c54d 100644
--- a/independent-projects/arc/runtime/src/main/java/io/quarkus/arc/impl/EventImpl.java
+++ b/independent-projects/arc/runtime/src/main/java/io/quarkus/arc/impl/EventImpl.java
@@ -207,9 +207,10 @@ private void handleExceptions(ObserverExceptionHandler handler) {
                 exception = new CompletionException(handledExceptions.get(0));
             } else {
                 exception = new CompletionException(null);
-                for (Throwable handledException : handledExceptions) {
-                    exception.addSuppressed(handledException);
-                }
+            }
+            // always add exceptions into suppressed
+            for (Throwable handledException : handledExceptions) {
+                exception.addSuppressed(handledException);
             }
             throw exception;
         }
diff --git a/independent-projects/arc/tests/src/test/java/io/quarkus/arc/test/observers/async/AsyncObserverExceptionTest.java b/independent-projects/arc/tests/src/test/java/io/quarkus/arc/test/observers/async/AsyncObserverExceptionTest.java
index ca43f015c20..781111b9ccf 100644
--- a/independent-projects/arc/tests/src/test/java/io/quarkus/arc/test/observers/async/AsyncObserverExceptionTest.java
+++ b/independent-projects/arc/tests/src/test/java/io/quarkus/arc/test/observers/async/AsyncObserverExceptionTest.java
@@ -2,19 +2,21 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import io.quarkus.arc.Arc;
 import io.quarkus.arc.ArcContainer;
 import io.quarkus.arc.test.ArcTestContainer;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import javax.annotation.PostConstruct;
 import javax.annotation.Priority;
 import javax.enterprise.context.Dependent;
@@ -31,20 +33,30 @@ public class AsyncObserverExceptionTest {
     public ArcTestContainer container = new ArcTestContainer(StringProducer.class, StringObserver.class);
 
     @Test
-    public void testAsyncObservers() throws InterruptedException, ExecutionException, TimeoutException {
+    public void testAsyncObserversSingleException() throws InterruptedException {
         ArcContainer container = Arc.container();
         StringProducer producer = container.instance(StringProducer.class).get();
         StringObserver observer = container.instance(StringObserver.class).get();
 
-        BlockingQueue<Object> synchronizer = new LinkedBlockingQueue<>();
+        final List<Throwable> suppressed = new ArrayList<>();
+        BlockingQueue<Throwable> synchronizer = new LinkedBlockingQueue<>();
         producer.produceAsync("pong").exceptionally(ex -> {
+            Arrays.stream(ex.getSuppressed()).forEach(t -> suppressed.add(t));
             synchronizer.add(ex);
             return ex.getMessage();
         });
 
-        Object exception = synchronizer.poll(10, TimeUnit.SECONDS);
+        Throwable exception = synchronizer.poll(10, TimeUnit.SECONDS);
+
+        // assert suppressed exception is always present
+        assertEquals(1, suppressed.size());
+        assertTrue(suppressed.get(0) instanceof RuntimeException);
+
+        // assert actual exception, always a CompletionException
         assertNotNull(exception);
-        assertTrue(exception instanceof RuntimeException);
+        assertTrue(exception instanceof CompletionException);
+        // in case of single exception in event chain, the cause is the exception
+        assertTrue(exception.getCause() instanceof RuntimeException);
 
         List<String> events = observer.getEvents();
         assertEquals(2, events.size());
@@ -52,6 +64,40 @@ public void testAsyncObservers() throws InterruptedException, ExecutionException
         assertEquals("async2::pong", events.get(1));
     }
 
+    @Test
+    public void testAsyncObserversMultipleExceptions() throws InterruptedException {
+        ArcContainer container = Arc.container();
+        StringProducer producer = container.instance(StringProducer.class).get();
+        StringObserver observer = container.instance(StringObserver.class).get();
+
+        final List<Throwable> suppressed = new ArrayList<>();
+        BlockingQueue<Throwable> synchronizer = new LinkedBlockingQueue<>();
+        producer.produceAsync("ping").exceptionally(ex -> {
+            Arrays.stream(ex.getSuppressed()).forEach(t -> suppressed.add(t));
+            synchronizer.add(ex);
+            return ex.getMessage();
+        });
+
+        Throwable exception = synchronizer.poll(10, TimeUnit.SECONDS);
+
+        // assert all suppressed exceptions are present
+        assertEquals(2, suppressed.size());
+        assertTrue(suppressed.get(0) instanceof RuntimeException);
+        assertTrue(suppressed.get(1) instanceof IllegalStateException);
+
+        // assert actual exception, always a CompletionException
+        assertNotNull(exception);
+        assertTrue(exception instanceof CompletionException);
+        // in case of multiple exceptions in event chain, the cause is expected to be null
+        assertNull(exception.getCause());
+
+        List<String> events = observer.getEvents();
+        assertEquals(3, events.size());
+        assertEquals("async1::ping", events.get(0));
+        assertEquals("async2::ping", events.get(1));
+        assertEquals("async3::ping", events.get(2));
+    }
+
     @Singleton
     static class StringObserver {
 
@@ -71,6 +117,13 @@ void observeAsync2(@ObservesAsync @Priority(2) String value) {
             events.add("async2::" + value);
         }
 
+        void observeAsync3(@ObservesAsync @Priority(3) String value) {
+            if (value.equals("ping")) {
+                events.add("async3::" + value);
+                throw new IllegalStateException("nok");
+            }
+        }
+
         List<String> getEvents() {
             return events;
         }