This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 8e4d1c2a0e84 CAMEL-25314: camel-cxf - a suspended consumer must reject
new requests (#27390)
8e4d1c2a0e84 is described below
commit 8e4d1c2a0e84e654007dccace7156e81f20902ff
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 6 14:50:41 2026 +0530
CAMEL-25314: camel-cxf - a suspended consumer must reject new requests
(#27390)
* CAMEL-25314: camel-cxf - a suspended consumer must reject new requests
CxfConsumer and CxfRsConsumer are Suspendable (CAMEL-12870), so that a
graceful shutdown suspends them and keeps the CXF server running while the
requests in flight complete. But they have no suspend logic, and their
invokers never check the status: a suspended consumer kept accepting and
routing new requests. A graceful shutdown with a steady flow of requests
therefore waited for the shutdown timeout (and then forced the requests in
flight), and suspending the route (route controller, JMX, route policies
such
as ThrottlingInflightRoutePolicy) did not stop it.
The invokers now reject a new invocation (synchronous, or a new
continuation)
with HTTP status 503 while the consumer is suspending or suspended, like the
HTTP consumers do. A resumed continuation belongs to a request in flight and
still completes.
The rejection is returned as a 503 response instead of a
WebApplicationException, which CXF's
WebApplicationExceptionMapper logged at WARN with its stack trace for every
rejected request. The
continuation is not looked up for the invocation of a sub resource locator
(as before): the method of the
sub resource is invoked next and is checked then, and getting the
continuation twice starts the async
context of the servlet request again. New test: a graceful stop rejects a
new request with 503 while the
asynchronous request in flight (continuation resumed after the suspend)
still completes.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25314: camel-cxf - a rejected request of a suspended consumer is
not logged as a CXF warning, and the test shuts down its bus
CXF logs a fault it does not expect at WARN level with its stack trace, so
a graceful shutdown under load logged one
stack trace per rejected request. The rejection now sets a FaultListener on
the request message that skips CXF's
default logging of that fault (the consumer logs the rejection at DEBUG);
the 503 response is unchanged.
Review nit: CxfConsumerSuspendTest reuses one bus for its clients and shuts
it down after each test.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/component/cxf/jaxrs/CxfRsInvoker.java | 19 +-
.../cxf/jaxrs/CxfRsConsumerSuspendTest.java | 113 ++++++++++++
.../camel/component/cxf/jaxws/CxfConsumer.java | 26 +++
.../cxf/jaxws/CxfConsumerSuspendTest.java | 195 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 10 ++
5 files changed, 360 insertions(+), 3 deletions(-)
diff --git
a/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
b/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
index 768d01be35c1..190285304cce 100644
---
a/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
+++
b/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
@@ -22,6 +22,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import jakarta.ws.rs.WebApplicationException;
import jakarta.ws.rs.core.HttpHeaders;
import jakarta.ws.rs.core.Request;
+import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.SecurityContext;
import jakarta.ws.rs.core.UriInfo;
@@ -58,18 +59,30 @@ public class CxfRsInvoker extends JAXRSInvoker {
Exchange cxfExchange, final Object serviceObject, Method method,
Object[] paramArray)
throws Exception {
+ OperationResourceInfo ori =
cxfExchange.get(OperationResourceInfo.class);
+ // a sub resource locator only returns the sub resource, whose method
is invoked next (and checked then)
+ Continuation continuation = null;
+ if (!ori.isSubResourceLocator()) {
+ continuation = endpoint.isSynchronous() ? null :
getContinuation(cxfExchange);
+ // a resumed continuation belongs to a request that is in flight
and must complete
+ if ((continuation == null || continuation.isNew()) &&
cxfRsConsumer.isSuspendingOrSuspended()) {
+ // a suspended consumer (suspended route, graceful shutdown)
does not accept new requests, like the
+ // HTTP consumers, which answer 503 (a response rather than an
exception, which CXF logs with its
+ // stack trace)
+ LOG.debug("Consumer suspended, cannot service request");
+ return
Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
+ }
+ }
Object response = null;
if (endpoint.isPerformInvocation()) {
response = super.performInvocation(cxfExchange, serviceObject,
method, paramArray);
}
paramArray = insertExchange(method, paramArray, cxfExchange);
- OperationResourceInfo ori =
cxfExchange.get(OperationResourceInfo.class);
if (ori.isSubResourceLocator()) {
// don't delegate the sub resource locator call to camel processor
return method.invoke(serviceObject, paramArray);
}
- Continuation continuation;
- if (!endpoint.isSynchronous() && (continuation =
getContinuation(cxfExchange)) != null) {
+ if (continuation != null) {
LOG.trace("Calling the Camel async processors.");
return asyncInvoke(cxfExchange, method, paramArray, continuation,
response);
} else {
diff --git
a/components/camel-cxf/camel-cxf-rest/src/test/java/org/apache/camel/component/cxf/jaxrs/CxfRsConsumerSuspendTest.java
b/components/camel-cxf/camel-cxf-rest/src/test/java/org/apache/camel/component/cxf/jaxrs/CxfRsConsumerSuspendTest.java
new file mode 100644
index 000000000000..4780414c2550
--- /dev/null
+++
b/components/camel-cxf/camel-cxf-rest/src/test/java/org/apache/camel/component/cxf/jaxrs/CxfRsConsumerSuspendTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.jaxrs;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.cxf.common.CXFTestSupport;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A suspended CXF-RS consumer (suspended route, graceful shutdown, route
policies) must not process new requests.
+ */
+class CxfRsConsumerSuspendTest extends CamelTestSupport {
+
+ private static final String CONTEXT = "/CxfRsConsumerSuspendTest";
+ private static final String URL = "http://localhost:"; +
CXFTestSupport.getPort1() + CONTEXT
+ + "/rest/customerservice/customers/123";
+
+ private final AtomicInteger processed = new AtomicInteger();
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private volatile boolean block;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("cxfrs://http://localhost:"; + CXFTestSupport.getPort1() +
CONTEXT + "/rest"
+ +
"?resourceClasses=org.apache.camel.component.cxf.jaxrs.testbean.CustomerService")
+ .routeId("rs")
+ .process(exchange -> {
+ if (processed.incrementAndGet() == 1 && block) {
+ entered.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ }
+ })
+ .setBody(constant("ok"));
+ }
+ };
+ }
+
+ @Test
+ void testSuspendedRouteRejectsRequests() throws Exception {
+ HttpClient client = HttpClient.newHttpClient();
+ HttpRequest request =
HttpRequest.newBuilder(URI.create(URL)).GET().build();
+ assertEquals(200, client.send(request,
HttpResponse.BodyHandlers.ofString()).statusCode());
+
+ context.getRouteController().suspendRoute("rs");
+ assertEquals(503, client.send(request,
HttpResponse.BodyHandlers.ofString()).statusCode(),
+ "A suspended route must answer 503");
+ assertEquals(1, processed.get(), "A suspended route must not process a
request");
+
+ context.getRouteController().resumeRoute("rs");
+ assertEquals(200, client.send(request,
HttpResponse.BodyHandlers.ofString()).statusCode());
+ assertEquals(2, processed.get());
+ }
+
+ @Test
+ void testGracefulStopCompletesInflightAndRejectsNewRequests() throws
Exception {
+ block = true;
+ HttpClient client = HttpClient.newHttpClient();
+ HttpRequest request =
HttpRequest.newBuilder(URI.create(URL)).GET().build();
+ // the consumer is asynchronous by default: the continuation of this
request resumes after the suspend
+ CompletableFuture<HttpResponse<String>> inflight =
client.sendAsync(request, HttpResponse.BodyHandlers.ofString());
+ assertTrue(entered.await(20, TimeUnit.SECONDS));
+
+ CompletableFuture<Void> stop = CompletableFuture.runAsync(() -> {
+ try {
+ context.getRouteController().stopRoute("rs", 20,
TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("rs").getConsumer();
+ await().atMost(20,
TimeUnit.SECONDS).until(consumer::isSuspendingOrSuspended);
+
+ assertEquals(503, client.send(request,
HttpResponse.BodyHandlers.ofString()).statusCode(),
+ "A request received while stopping must be rejected");
+ assertEquals(1, processed.get(), "A request received while stopping
must not be processed");
+
+ release.countDown();
+ assertEquals(200, inflight.get(20, TimeUnit.SECONDS).statusCode());
+ stop.get(30, TimeUnit.SECONDS);
+ }
+}
diff --git
a/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java
b/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java
index b636b78b3726..00b1da356367 100644
---
a/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java
+++
b/components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java
@@ -48,6 +48,7 @@ import org.apache.cxf.continuations.ContinuationProvider;
import org.apache.cxf.endpoint.Server;
import org.apache.cxf.frontend.ServerFactoryBean;
import org.apache.cxf.interceptor.Fault;
+import org.apache.cxf.logging.FaultListener;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.FaultMode;
import org.apache.cxf.message.Message;
@@ -67,6 +68,8 @@ import org.slf4j.LoggerFactory;
public class CxfConsumer extends DefaultConsumer implements Suspendable {
private static final Logger LOG =
LoggerFactory.getLogger(CxfConsumer.class);
+ // skips the default logging of a fault by CXF
+ private static final FaultListener NO_FAULT_LOGGING = (exception,
description, message) -> false;
private Server server;
private CxfEndpoint cxfEndpoint;
@@ -168,14 +171,37 @@ public class CxfConsumer extends DefaultConsumer
implements Suspendable {
Continuation continuation;
if (!endpoint.isSynchronous() &&
isAsyncInvocationSupported(cxfExchange)
&& (continuation = getContinuation(cxfExchange)) != null) {
+ // a resumed continuation belongs to a request that is in
flight and must complete
+ if (continuation.isNew()) {
+ rejectIfSuspended(cxfExchange);
+ }
LOG.trace("Calling the Camel async processors.");
return asyncInvoke(cxfExchange, continuation);
} else {
+ rejectIfSuspended(cxfExchange);
LOG.trace("Calling the Camel sync processors.");
return syncInvoke(cxfExchange);
}
}
+ // a suspended consumer (suspended route, graceful shutdown) does not
accept new requests, like the HTTP
+ // consumers, which answer 503
+ private void rejectIfSuspended(Exchange cxfExchange) {
+ if (isSuspendingOrSuspended()) {
+ LOG.debug("Consumer suspended, cannot service request");
+ // CXF logs a fault it does not expect at WARN level with its
stack trace (PhaseInterceptorChain),
+ // unless a FaultListener of the message says otherwise: a
rejected request is expected while
+ // suspended (e.g. every request during a graceful shutdown
under load), so it is only logged above
+ Message in = cxfExchange.getInMessage();
+ if (in != null) {
+ in.put(FaultListener.class.getName(), NO_FAULT_LOGGING);
+ }
+ Fault fault = new Fault(new IllegalStateException("Service
unavailable: the consumer is suspended"));
+ fault.setStatusCode(503);
+ throw fault;
+ }
+ }
+
// NOTE this code cannot work with CXF 2.2.x and JMSContinuation
// as it doesn't break out the interceptor chain when we call it
private Object asyncInvoke(Exchange cxfExchange, final Continuation
continuation) {
diff --git
a/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerSuspendTest.java
b/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerSuspendTest.java
new file mode 100644
index 000000000000..5f22a2e74342
--- /dev/null
+++
b/components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerSuspendTest.java
@@ -0,0 +1,195 @@
+/*
+ * 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 java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.cxf.common.CXFTestSupport;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.frontend.ClientFactoryBean;
+import org.apache.cxf.frontend.ClientProxyFactoryBean;
+import org.apache.cxf.transport.http.HTTPException;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A suspended CXF consumer (suspended route, graceful shutdown, route
policies) must not process new requests, while
+ * the requests in flight complete.
+ */
+class CxfConsumerSuspendTest extends CamelTestSupport {
+
+ private static final String ADDRESS = "http://localhost:"; +
CXFTestSupport.getPort1() + "/CxfConsumerSuspendTest/test";
+
+ private final AtomicInteger processed = new AtomicInteger();
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private volatile boolean block;
+ // the bus of the test clients
+ private Bus clientBus;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("cxf://" + ADDRESS +
"?serviceClass=org.apache.camel.component.cxf.jaxws.HelloService")
+ .routeId("cxf")
+ .process(exchange -> {
+ if (processed.incrementAndGet() == 1 && block) {
+ entered.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ }
+ String text = (String)
exchange.getIn().getBody(List.class).get(0);
+ exchange.getMessage().setBody("echo " + text);
+ });
+ }
+ };
+ }
+
+ @Test
+ void testSuspendedRouteRejectsRequests() throws Exception {
+ HelloService client = createClient();
+ assertEquals("echo a", client.echo("a"));
+
+ context.getRouteController().suspendRoute("cxf");
+ Exception e = assertThrows(Exception.class, () -> client.echo("b"), "A
suspended route must not accept a request");
+ assertServiceUnavailable(e);
+ assertEquals(1, processed.get(), "A suspended route must not process a
request");
+
+ context.getRouteController().resumeRoute("cxf");
+ assertEquals("echo c", client.echo("c"));
+ assertEquals(2, processed.get());
+ }
+
+ @Test
+ void testRejectedRequestIsNotLoggedAsWarning() throws Exception {
+ HelloService client = createClient();
+ assertEquals("echo a", client.echo("a"));
+ context.getRouteController().suspendRoute("cxf");
+
+ // CXF logs an unexpected fault at WARN level with its stack trace: a
rejected request is expected while the
+ // consumer is suspended (e.g. every request during a graceful
shutdown under load)
+ // (the client, on this thread, logs the 503 fault it receives)
+ String clientThread = Thread.currentThread().getName();
+ List<LogEvent> warnings = new CopyOnWriteArrayList<>();
+ AbstractAppender appender = new
AbstractAppender("CxfConsumerSuspendTest", null, null, true,
Property.EMPTY_ARRAY) {
+ @Override
+ public void append(LogEvent event) {
+ if (event.getLevel().isMoreSpecificThan(Level.WARN) &&
event.getLoggerName().startsWith("org.apache.cxf")
+ && !clientThread.equals(event.getThreadName())) {
+ warnings.add(event.toImmutable());
+ }
+ }
+ };
+ appender.start();
+ Logger cxfLogger = (Logger) LogManager.getLogger("org.apache.cxf");
+ cxfLogger.addAppender(appender);
+ try {
+ Exception e = assertThrows(Exception.class, () ->
client.echo("b"));
+ assertServiceUnavailable(e);
+ } finally {
+ cxfLogger.removeAppender(appender);
+ appender.stop();
+ }
+ assertTrue(warnings.isEmpty(), () -> "A rejected request must not be
logged as a warning by the CXF server: "
+ + warnings.stream()
+ .map(w ->
w.getThreadName() + " " + w.getLoggerName() + ": "
+ +
w.getMessage().getFormattedMessage())
+ .toList());
+ }
+
+ @Test
+ void testGracefulStopCompletesInflightAndRejectsNewRequests() throws
Exception {
+ block = true;
+ HelloService client = createClient();
+ CompletableFuture<String> inflight = CompletableFuture.supplyAsync(()
-> {
+ try {
+ return client.echo("a");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ assertTrue(entered.await(20, TimeUnit.SECONDS));
+
+ CompletableFuture<Void> stop = CompletableFuture.runAsync(() -> {
+ try {
+ context.getRouteController().stopRoute("cxf", 20,
TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("cxf").getConsumer();
+ await().atMost(20,
TimeUnit.SECONDS).until(consumer::isSuspendingOrSuspended);
+
+ HelloService other = createClient();
+ Exception e = assertThrows(Exception.class, () -> other.echo("b"),
+ "A request received while stopping must be rejected");
+ assertServiceUnavailable(e);
+ assertEquals(1, processed.get(), "A request received while stopping
must not be processed");
+
+ release.countDown();
+ assertEquals("echo a", inflight.get(20, TimeUnit.SECONDS));
+ stop.get(30, TimeUnit.SECONDS);
+ }
+
+ private static void assertServiceUnavailable(Exception e) {
+ HTTPException http = ObjectHelper.getException(HTTPException.class, e);
+ assertNotNull(http, "Expected an HTTP 503 response, got: " + e);
+ assertEquals(503, http.getResponseCode());
+ }
+
+ private HelloService createClient() {
+ if (clientBus == null) {
+ clientBus = BusFactory.newInstance().createBus();
+ }
+ ClientProxyFactoryBean proxyFactory = new ClientProxyFactoryBean();
+ ClientFactoryBean clientBean = proxyFactory.getClientFactoryBean();
+ clientBean.setAddress(ADDRESS);
+ clientBean.setServiceClass(HelloService.class);
+ clientBean.setBus(clientBus);
+ return (HelloService) proxyFactory.create();
+ }
+
+ @AfterEach
+ void shutdownClientBus() {
+ if (clientBus != null) {
+ clientBus.shutdown(true);
+ clientBus = null;
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 12fc32e9a85e..d254cf412111 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -2593,6 +2593,16 @@ A route that relies on an undeclared exception's message
reaching the caller mus
cxf://http://localhost:8080/service?serviceClass=com.example.MyService&muteException=false
----
+==== A suspended consumer rejects new requests
+
+The CXF consumers (`cxf` and `cxfrs`) are suspendable, so that a graceful
shutdown keeps the CXF server running
+while the requests in flight complete, but a suspended consumer kept accepting
and routing new requests: during a
+graceful shutdown (until the shutdown timeout), and while the route was
suspended (route controller, JMX, route
+policies such as `ThrottlingInflightRoutePolicy`). A suspended consumer now
rejects a new request with HTTP status
+503 (Service Unavailable), like the HTTP consumers do; the requests in flight,
including asynchronous ones whose
+continuation resumes later, still complete. The `cxf` consumer rejects the
request with a SOAP fault, which is
+logged at DEBUG level only; over a transport other than HTTP the caller gets
the fault only.
+
=== camel-thrift
The `thrift` data format used to deserialize into its own `defaultInstance`
and return that object to
