CAMEL-24476/24477/24478: muteException for the mina, cxf and grpc consumers, backport to camel-4.22.x

components/camel-cxf/camel-cxf-soap/pom.xml

@@ -210,6 +210,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>

components/camel-cxf/camel-cxf-soap/src/generated/java/org/apache/camel/component/cxf/jaxws/CxfEndpointConfigurer.java

@@ -64,6 +64,8 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj
         case "mergeProtocolHeaders": target.setMergeProtocolHeaders(property(camelContext, boolean.class, value)); return true;
         case "mtomenabled":
         case "mtomEnabled": target.setMtomEnabled(property(camelContext, boolean.class, value)); return true;
+        case "muteexception":
+        case "muteException": target.setMuteException(property(camelContext, boolean.class, value)); return true;
         case "password": target.setPassword(property(camelContext, java.lang.String.class, value)); return true;
         case "portname":
         case "portName": target.setPortName(property(camelContext, java.lang.String.class, value)); return true;
@@ -137,6 +139,8 @@ public Class<?> getOptionType(String name, boolean ignoreCase) {
         case "mergeProtocolHeaders": return boolean.class;
         case "mtomenabled":
         case "mtomEnabled": return boolean.class;
+        case "muteexception":
+        case "muteException": return boolean.class;
         case "password": return java.lang.String.class;
         case "portname":
         case "portName": return java.lang.String.class;
@@ -211,6 +215,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
         case "mergeProtocolHeaders": return target.isMergeProtocolHeaders();
         case "mtomenabled":
         case "mtomEnabled": return target.isMtomEnabled();
+        case "muteexception":
+        case "muteException": return target.isMuteException();
         case "password": return target.getPassword();
         case "portname":
         case "portName": return target.getPortName();

components/camel-cxf/camel-cxf-soap/src/generated/java/org/apache/camel/component/cxf/jaxws/CxfEndpointUriFactory.java

@@ -24,7 +24,7 @@ public class CxfEndpointUriFactory extends org.apache.camel.support.component.En
     private static final Set<String> ENDPOINT_IDENTITY_PROPERTY_NAMES;
     private static final Map<String, String> MULTI_VALUE_PREFIXES;
     static {
-        Set<String> props = new HashSet<>(38);
+        Set<String> props = new HashSet<>(39);
         props.add("address");
         props.add("allowStreaming");
         props.add("beanId");
@@ -48,6 +48,7 @@ public class CxfEndpointUriFactory extends org.apache.camel.support.component.En
         props.add("loggingSizeLimit");
         props.add("mergeProtocolHeaders");
         props.add("mtomEnabled");
+        props.add("muteException");
         props.add("password");
         props.add("portName");
         props.add("properties");

components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java

@@ -153,6 +153,7 @@ protected boolean isAsyncInvocationSupported(Exchange cxfExchange) {
     private class CxfConsumerInvoker implements Invoker {
 
         private static final String COMPLETED = "org.apache.camel.component.cxf.jaxws.completed";
+        private static final String MUTED_FAULT_MESSAGE = "Exchange processing failed";
 
         private final CxfEndpoint endpoint;
 
@@ -359,12 +360,17 @@ private void setResponseBack(Exchange cxfExchange, org.apache.camel.Exchange cam
         }
 
         private void checkFailure(org.apache.camel.Exchange camelExchange, Exchange cxfExchange) throws Fault {
-            final Throwable t = extractThrowable(camelExchange);
+            // an exception set on the exchange is a route failure; one carried in the body is a fault the route
+            // chose to return, and stays part of the contract either way
+            final Throwable routeFailure = camelExchange.getException();
+            final Throwable t = routeFailure != null ? routeFailure : extractFromBody(camelExchange, null);
 
             if (t != null) {
                 cxfExchange.getInMessage().put(FaultMode.class, FaultMode.UNCHECKED_APPLICATION_FAULT);
                 if (t instanceof Fault fault) {
                     handleFault(cxfExchange, fault);
+                } else if (routeFailure != null && isMuted(t)) {
+                    buildFaultFromThrowable(mutedThrowable());
                 } else {
                     // This is not a CXF Fault. Build the CXF Fault manually.
                     buildFaultFromThrowable(t);
@@ -373,6 +379,26 @@ private void checkFailure(org.apache.camel.Exchange camelExchange, Exchange cxfE
             }
         }
 
+        /**
+         * Only an <em>undeclared</em> route failure is muted. An exception the service contract declares - annotated
+         * {@code @WebFault} - is what a SOAP client is written against, so suppressing it would break the contract
+         * rather than protect anything.
+         */
+        private boolean isMuted(Throwable t) {
+            return ((CxfEndpoint) getEndpoint()).isMuteException()
+                    && t.getClass().getAnnotation(WebFault.class) == null;
+        }
+
+        /**
+         * Carries no class name, no message and no stack trace of the route's exception: CXF copies the message into
+         * the SOAP faultstring, and can be configured to put the stack trace in the fault detail.
+         */
+        private static Throwable mutedThrowable() {
+            Exception muted = new Exception(MUTED_FAULT_MESSAGE);
+            muted.setStackTrace(new StackTraceElement[0]);
+            return muted;
+        }
+
         private static void buildFaultFromThrowable(Throwable t) {
             Fault fault = new Fault(t);
             if (fault.getMessage() == null) {
@@ -410,15 +436,9 @@ private static void handleFault(Exchange cxfExchange, Fault t) {
             throw t;
         }
 
-        private static Throwable extractThrowable(org.apache.camel.Exchange camelExchange) {
-            Throwable t = camelExchange.getException();
-            if (t == null) {
-                // SOAP faults can be stored as exceptions as message body (to be backwards compatible)
-                t = extractFromBody(camelExchange, t);
-            }
-            return t;
-        }
-
+        /**
+         * SOAP faults can be stored as exceptions as message body (to be backwards compatible).
+         */
         private static Throwable extractFromBody(org.apache.camel.Exchange camelExchange, Throwable t) {
             Object body = camelExchange.getMessage().getBody();
             if (body instanceof Throwable throwable) {

components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfEndpoint.java

@@ -203,6 +203,8 @@ public class CxfEndpoint extends DefaultEndpoint implements AsyncEndpoint, Heade
     private boolean skipPayloadMessagePartCheck;
     @UriParam(label = "logging")
     private boolean skipFaultLogging;
+    @UriParam(label = "consumer", defaultValue = "true")
+    private boolean muteException = true;
     @UriParam(label = "advanced")
     private boolean mergeProtocolHeaders;
     @UriParam(label = "advanced")
@@ -1458,6 +1460,20 @@ public void setSkipFaultLogging(boolean skipFaultLogging) {
         this.skipFaultLogging = skipFaultLogging;
     }
 
+    public boolean isMuteException() {
+        return muteException;
+    }
+
+    /**
+     * If enabled and an Exchange failed processing on the consumer side, and the exception is not a SOAP fault the
+     * service contract declares, the SOAP fault returned to the caller won't contain the exception's class or message.
+     * Faults the route raises deliberately - a CXF {@code Fault}, an exception annotated {@code @WebFault}, or a
+     * {@code Throwable} set as the message body - are part of the contract and are always returned in full.
+     */
+    public void setMuteException(boolean muteException) {
+        this.muteException = muteException;
+    }
+
     public boolean isMergeProtocolHeaders() {
         return mergeProtocolHeaders;
     }

components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerContinuationTimeoutTest.java

@@ -62,42 +62,45 @@ public void configure() {
                         .setBody(constant("Sensitive Data"))
                         .to(simpleEndpointURI + "&continuationTimeout=5000&dataFormat=RAW");
 
-                from(simpleEndpointURI + "&continuationTimeout=5000&dataFormat=RAW").process(new AsyncProcessorSupport() {
-                    @Override
-                    public boolean process(Exchange exchange, AsyncCallback asyncCallback) {
-                        Message in = exchange.getIn();
-                        // check the content-length header is filtered
-                        Object value = in.getHeader("Content-Length");
-                        assertNull(value, "The Content-Length header should be removed");
-                        // Get the request message
-                        String request = in.getBody(String.class);
-                        String priority = in.getHeader("priority", "fast", String.class);
-
-                        // need not to block this thread to simulate slow response so use a thread pool to wait
-                        if ("slow".equalsIgnoreCase(priority)) {
-                            pool.submit(() -> {
-                                try {
-                                    log.info("Sleeping for 10 seconds to simulate slow response");
-                                    Thread.sleep(10000);
-                                } catch (InterruptedException e) {
-                                    // ignore
-                                } finally {
-                                    asyncCallback.done(false);
+                // muteException=false: these tests assert the continuation-timeout message reaches the caller,
+                // which CAMEL-24477 otherwise replaces with a generic fault
+                from(simpleEndpointURI + "&continuationTimeout=5000&dataFormat=RAW&muteException=false")
+                        .process(new AsyncProcessorSupport() {
+                            @Override
+                            public boolean process(Exchange exchange, AsyncCallback asyncCallback) {
+                                Message in = exchange.getIn();
+                                // check the content-length header is filtered
+                                Object value = in.getHeader("Content-Length");
+                                assertNull(value, "The Content-Length header should be removed");
+                                // Get the request message
+                                String request = in.getBody(String.class);
+                                String priority = in.getHeader("priority", "fast", String.class);
+
+                                // need not to block this thread to simulate slow response so use a thread pool to wait
+                                if ("slow".equalsIgnoreCase(priority)) {
+                                    pool.submit(() -> {
+                                        try {
+                                            log.info("Sleeping for 10 seconds to simulate slow response");
+                                            Thread.sleep(10000);
+                                        } catch (InterruptedException e) {
+                                            // ignore
+                                        } finally {
+                                            asyncCallback.done(false);
+                                        }
+                                    });
+                                    return false;
+                                } else {
+                                    // Send the response message back
+                                    if (request.indexOf(ECHO_METHOD) > 0) {
+                                        exchange.getMessage().setBody(ECHO_RESPONSE);
+                                    } else { // echoBoolean call
+                                        exchange.getMessage().setBody(ECHO_BOOLEAN_RESPONSE);
+                                    }
                                 }
-                            });
-                            return false;
-                        } else {
-                            // Send the response message back
-                            if (request.indexOf(ECHO_METHOD) > 0) {
-                                exchange.getMessage().setBody(ECHO_RESPONSE);
-                            } else { // echoBoolean call
-                                exchange.getMessage().setBody(ECHO_BOOLEAN_RESPONSE);
+                                asyncCallback.done(true);
+                                return true;
                             }
-                        }
-                        asyncCallback.done(true);
-                        return true;
-                    }
-                });
+                        });
             }
         };
     }

components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerMuteExceptionTest.java

@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.cxf.jaxws;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.cxf.common.CXFTestSupport;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.cxf.frontend.ClientFactoryBean;
+import org.apache.cxf.greeter_control.Greeter;
+import org.apache.cxf.greeter_control.PingMeFault;
+import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * The consumer used to put the route exception's message into the SOAP fault returned to the caller, so any
+ * remote-triggerable failure handed the caller internal detail.
+ * <p>
+ * {@code muteException} defaults to true, matching the http consumers aligned by CAMEL-23651. It applies only to
+ * <em>undeclared</em> failures: an exception the service contract declares with {@code @WebFault} is what a SOAP client
+ * is written against, so it is still returned in full.
+ */
+class CxfConsumerMuteExceptionTest extends CamelTestSupport {
+
+    private static final String DETAIL = "the-internal-detail-a-caller-must-not-see";
+
+    private static final String MUTED_ADDRESS
+            = "http://localhost:" + CXFTestSupport.getPort1() + "/CxfConsumerMuteExceptionTest/muted";
+    private static final String UNMUTED_ADDRESS
+            = "http://localhost:" + CXFTestSupport.getPort2() + "/CxfConsumerMuteExceptionTest/unmuted";
+    private static final String DECLARED_ADDRESS
+            = "http://localhost:" + CXFTestSupport.getPort3() + "/CxfConsumerMuteExceptionTest/declared";
+
+    private static final String SERVICE_CLASS = "serviceClass=org.apache.cxf.greeter_control.Greeter";
+
+    @Test
+    void anUndeclaredFailureIsNotDescribedToTheCaller() {
+        assertThatThrownBy(() -> client(MUTED_ADDRESS).pingMe())
+                .as("the SOAP fault must carry neither the class nor the message of the route's exception")
+                .hasMessageNotContaining(DETAIL)
+                .hasMessageNotContaining("IllegalStateException");
+    }
+
+    @Test
+    void muteExceptionFalseDescribesTheFailureAsBefore() {
+        assertThatThrownBy(() -> client(UNMUTED_ADDRESS).pingMe())
+                .hasMessageContaining(DETAIL);
+    }
+
+    /**
+     * The guarantee that makes muting safe to default on: a fault the WSDL declares is part of the contract, so muting
+     * must not touch it. Without this carve-out every SOAP client written against a declared fault would break.
+     */
+    @Test
+    void aDeclaredWebFaultIsStillReturnedInFullWhileMuted() {
+        assertThatThrownBy(() -> client(DECLARED_ADDRESS).pingMe())
+                .isInstanceOf(PingMeFault.class)
+                .hasMessageContaining(DETAIL);
+    }
+
+    @Test
+    void muteExceptionDefaultsToTrue() {
+        CxfEndpoint endpoint = context.getEndpoint("cxf://" + MUTED_ADDRESS + "?" + SERVICE_CLASS, CxfEndpoint.class);
+        assertThat(endpoint.isMuteException()).isTrue();
+    }
+
+    private static Greeter client(String address) {
+        JaxWsProxyFactoryBean proxyFactory = new JaxWsProxyFactoryBean();
+        ClientFactoryBean clientBean = proxyFactory.getClientFactoryBean();
+        clientBean.setAddress(address);
+        clientBean.setServiceClass(Greeter.class);
+        return (Greeter) proxyFactory.create();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("cxf://" + MUTED_ADDRESS + "?" + SERVICE_CLASS)
+                        .process(e -> {
+                            throw new IllegalStateException(DETAIL);
+                        });
+                from("cxf://" + UNMUTED_ADDRESS + "?" + SERVICE_CLASS + "&muteException=false")
+                        .process(e -> {
+                            throw new IllegalStateException(DETAIL);
+                        });
+                from("cxf://" + DECLARED_ADDRESS + "?" + SERVICE_CLASS)
+                        .process(e -> {
+                            throw new PingMeFault(DETAIL);
+                        });
+            }
+        };
+    }
+}

components/camel-grpc/src/generated/java/org/apache/camel/component/grpc/GrpcEndpointConfigurer.java

@@ -84,6 +84,8 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj
         case "maxrstperiodseconds":
         case "maxRstPeriodSeconds": target.getConfiguration().setMaxRstPeriodSeconds(property(camelContext, int.class, value)); return true;
         case "method": target.getConfiguration().setMethod(property(camelContext, java.lang.String.class, value)); return true;
+        case "muteexception":
+        case "muteException": target.getConfiguration().setMuteException(property(camelContext, boolean.class, value)); return true;
         case "negotiationtype":
         case "negotiationType": target.getConfiguration().setNegotiationType(property(camelContext, io.grpc.netty.NegotiationType.class, value)); return true;
         case "permitkeepalivetime":
@@ -173,6 +175,8 @@ public Class<?> getOptionType(String name, boolean ignoreCase) {
         case "maxrstperiodseconds":
         case "maxRstPeriodSeconds": return int.class;
         case "method": return java.lang.String.class;
+        case "muteexception":
+        case "muteException": return boolean.class;
         case "negotiationtype":
         case "negotiationType": return io.grpc.netty.NegotiationType.class;
         case "permitkeepalivetime":
@@ -263,6 +267,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
         case "maxrstperiodseconds":
         case "maxRstPeriodSeconds": return target.getConfiguration().getMaxRstPeriodSeconds();
         case "method": return target.getConfiguration().getMethod();
+        case "muteexception":
+        case "muteException": return target.getConfiguration().isMuteException();
         case "negotiationtype":
         case "negotiationType": return target.getConfiguration().getNegotiationType();
         case "permitkeepalivetime":