[CXF-9189]Exceptions Not Logged When Using SseEventSink in JAX-RS SSE… (#2919)

* [CXF-9189]Exceptions Not Logged When Using SseEventSink in JAX-RS SSE Endpoint

* [CXF-9189]use in memory log appender for the testcase

Author: ffang

rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/JAXRSInvoker.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import jakarta.ws.rs.WebApplicationException;
@@ -38,6 +39,7 @@
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.MultivaluedMap;
 import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.sse.SseEventSink;
 import org.apache.cxf.common.classloader.ClassLoaderUtils;
 import org.apache.cxf.common.classloader.ClassLoaderUtils.ClassLoaderHolder;
 import org.apache.cxf.common.i18n.BundleUtils;
@@ -144,6 +146,21 @@ private Object handleAsyncFault(Exchange exchange, AsyncResponseImpl ar, Throwab
         try {
             return handleFault(new Fault(t), exchange.getInMessage(), null, null);
         } catch (Fault ex) {
+            //If we got here, the fault was effectively "unmapped" (no Response could be created)
+            // and we'd otherwise lose the usual fault logging (common with SSE sink / async paths).
+            Throwable cause = ex.getCause() == null ? ex : ex.getCause();
+            LOG.log(Level.SEVERE, "Unhandled exception from JAX-RS invocation (async/SSE path)", cause);
+            
+            // Best-effort: if this request is SSE, close the sink so the container can complete cleanly.
+            try {
+                SseEventSink sink = (SseEventSink)exchange.getInMessage().get(SseEventSink.class);
+                if (sink != null && !sink.isClosed()) {
+                    sink.close();
+                }
+            } catch (Exception ignore) {
+                // ignore secondary failures; primary goal is to log original cause
+            }
+            
             ar.setUnmappedThrowable(ex.getCause() == null ? ex : ex.getCause());
             if (isSuspended(exchange)) {
                 ar.reset();

systests/rs-sse/rs-sse-base/pom.xml

@@ -75,5 +75,9 @@
             <artifactId>awaitility</artifactId>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
     </dependencies>
 </project>

systests/rs-sse/rs-sse-base/src/main/java/org/apache/cxf/systest/jaxrs/sse/AbstractSseTest.java

@@ -32,7 +32,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
-
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.IThrowableProxy;
+import ch.qos.logback.core.read.ListAppender;
 import jakarta.ws.rs.client.Entity;
 import jakarta.ws.rs.client.WebTarget;
 import jakarta.ws.rs.core.HttpHeaders;
@@ -43,9 +47,11 @@
 import jakarta.ws.rs.sse.InboundSseEvent;
 import jakarta.ws.rs.sse.SseEventSource;
 import jakarta.ws.rs.sse.SseEventSource.Builder;
+import org.slf4j.LoggerFactory;
 import tools.jackson.core.JacksonException;
 import tools.jackson.jakarta.rs.json.JacksonJsonProvider;
 
+
 import org.junit.Before;
 import org.junit.Test;
 
@@ -60,6 +66,10 @@
 import static org.junit.Assert.fail;
 
 
+
+
+
+
 public abstract class AbstractSseTest extends AbstractSseBaseTest {
     @Before
     public void setUp() {
@@ -70,6 +80,9 @@ public void setUp() {
 
     }
 
+    
+
+        
     @Test
     public void testBooksStreamIsReturnedFromLastEventId() throws InterruptedException {
         final WebTarget target = createWebTarget("/rest/api/bookstore/sse/" + UUID.randomUUID())
@@ -408,7 +421,98 @@ public void testBooksSseContainerResponseAddedHeaders() throws InterruptedExcept
             assertThat(response.getHeaderString("X-My-ProtocolHeader"), equalTo("protocol-headers"));
         }
     }
+    
+ 
+    @Test
+    public void testSseEndpointExceptionIsLoggedToConsole() throws Exception {
+        final Logger logger = (Logger) LoggerFactory.getLogger("org.apache.cxf.jaxrs.JAXRSInvoker");
+
+        final Level oldLevel = logger.getLevel();
+        final ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.start();
+
+        try {
+            logger.setLevel(Level.ERROR);
+            logger.addAppender(appender);
+
+            try (Response r = createWebTarget("/rest/api/bookstore/sse/fail/request")
+                    .request(MediaType.SERVER_SENT_EVENTS)
+                    .get()) {
+                // Force the client to actually start consuming
+                r.readEntity(String.class);
+            } catch (Exception ex) {
+                // expected
+            }
+
+            // Wait until we have at least one ERROR from JAXRSInvoker
+            awaitEvents(2000, appender.list, 1);
+
+            assertTrue("Expected SSE log event, got:\n" + dump(appender),
+                    hasUnhandledExceptionEvent(appender));
+
+            assertTrue("Expected SSE marker in throwable, got:\n" + dump(appender),
+                    hasMarkerInUnhandledExceptionEvent(appender, "CXF-9189-MARKER"));
+        } finally {
+            logger.detachAppender(appender);
+            logger.setLevel(oldLevel);
+            appender.stop();
+        }
+    }
 
+    private static boolean hasUnhandledExceptionEvent(ListAppender<ILoggingEvent> appender) {
+        final String msgNeedle = "Unhandled exception from JAX-RS invocation (async/SSE path)";
+        for (ILoggingEvent e : appender.list) {
+            String msg = e.getFormattedMessage();
+            if (msg != null && msg.contains(msgNeedle)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static boolean hasMarkerInUnhandledExceptionEvent(ListAppender<ILoggingEvent> appender, String marker) {
+        final String msgNeedle = "Unhandled exception from JAX-RS invocation (async/SSE path)";
+        for (ILoggingEvent e : appender.list) {
+            String msg = e.getFormattedMessage();
+            if (msg == null || !msg.contains(msgNeedle)) {
+                continue;
+            }
+            // marker can be in message OR in throwable chain
+            if (msg.contains(marker) || throwableChainContains(e.getThrowableProxy(), marker)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static boolean throwableChainContains(IThrowableProxy tp, String needle) {
+        for (IThrowableProxy cur = tp; cur != null; cur = cur.getCause()) {
+            String m = cur.getMessage();
+            if (m != null && m.contains(needle)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static String dump(ListAppender<ILoggingEvent> appender) {
+        StringBuilder sb = new StringBuilder();
+        for (ILoggingEvent e : appender.list) {
+            sb.append('[').append(e.getLevel()).append("] ")
+              .append(e.getLoggerName()).append(" - ")
+              .append(e.getFormattedMessage());
+            if (e.getThrowableProxy() != null) {
+                sb.append(" (thrown: ")
+                  .append(e.getThrowableProxy().getClassName())
+                  .append(": ")
+                  .append(e.getThrowableProxy().getMessage())
+                  .append(')');
+            }
+            sb.append('\n');
+        }
+        return sb.toString();
+    }
+    
     /**
      * Jetty / Undertow do not propagate errors from the runnable passed to
      * AsyncContext::start() up to the AsyncEventListener::onError(). Tomcat however
@@ -426,4 +530,6 @@ private static Consumer<InboundSseEvent> collect(final Collection<Book> books) {
     private static Consumer<InboundSseEvent> collectRaw(final Collection<String> titles) {
         return event -> titles.add(event.readData(String.class, MediaType.TEXT_PLAIN_TYPE));
     }
+    
+    
 }

systests/rs-sse/rs-sse-base/src/main/java/org/apache/cxf/systest/jaxrs/sse/BookStore.java

@@ -258,6 +258,15 @@ public int filteredStats() {
         return BookStoreResponseFilter.getInvocations();
     }
 
+    @GET
+    @Path("/sse/fail/request")
+    @Produces(MediaType.SERVER_SENT_EVENTS)
+    public void failOnRequestThread(@Context SseEventSink sink) {
+        throw new RuntimeException("CXF-9189-MARKER: exception from SSE resource method should be logged");
+    }
+
+    
+        
     @PUT
     @Path("/filtered/stats")
     public void clearStats() {

systests/rs-sse/rs-sse-base/src/main/java/org/apache/cxf/systest/jaxrs/sse/BookStore2.java

@@ -238,6 +238,16 @@ public void run() {
     public int filteredStats() {
         return BookStoreResponseFilter.getInvocations();
     }
+    
+    @GET
+    @Path("/sse/fail/request")
+    @Produces(MediaType.SERVER_SENT_EVENTS)
+    public void failOnRequestThread(@Context SseEventSink sink) {
+        throw new RuntimeException("CXF-9189-MARKER: exception from SSE resource method should be logged");
+    }
+
+    
+    
 
     @PUT
     @Path("/filtered/stats")