allthingssecurity commented on code in PR #27390:
URL: https://github.com/apache/camel/pull/27390#discussion_r4192127804
##########
components/camel-cxf/camel-cxf-soap/src/main/java/org/apache/camel/component/cxf/jaxws/CxfConsumer.java:
##########
@@ -168,14 +168,30 @@ public Object invoke(Exchange cxfExchange, Object o) {
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();
+ }
LOG.trace("Calling the Camel async processors.");
return asyncInvoke(cxfExchange, continuation);
} else {
+ rejectIfSuspended();
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() {
Review Comment:
Thanks for the review. Yes, there is a clean way, done in d8a1a7bbcaa9.
`PhaseInterceptorChain` logs a fault through `doDefaultLogging` (WARN with the
stack trace, unless `FaultMode` is `CHECKED_APPLICATION_FAULT`, then INFO
without it) only if the `FaultListener` found as a contextual property of the
message returns true. The rejection now puts a `FaultListener` that returns
false on that request's in-message, so CXF skips its log for this fault only,
and the consumer logs the rejection at DEBUG. The 503 status and the fault sent
to the caller are unchanged. I did not use `CHECKED_APPLICATION_FAULT`, as it
would mislabel the fault and also drives the declared-fault mapping.
One side effect: a user-configured `FaultListener` (bus or endpoint
property) is not called for these rejected requests. New
`testRejectedRequestIsNotLoggedAsWarning` captures WARN events of
`org.apache.cxf` loggers on the server threads: without the change it gets
`PhaseInterceptorChain: Interceptor for {...}HelloService#{...}echo has thrown
exception, unwinding now`; with it, none. The upgrade-guide note now says DEBUG.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-cxf/camel-cxf-soap/src/test/java/org/apache/camel/component/cxf/jaxws/CxfConsumerSuspendTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.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.BusFactory;
+import org.apache.cxf.frontend.ClientFactoryBean;
+import org.apache.cxf.frontend.ClientProxyFactoryBean;
+import org.apache.cxf.transport.http.HTTPException;
+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;
+
+ @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();
Review Comment:
Fixed in d8a1a7bbcaa9: the test creates one client bus lazily, reuses it for
every client of the test, and shuts it down in an `@AfterEach`. camel-cxf-soap:
169 tests pass (2 skipped); camel-cxf-rest, camel-cxf-spring-soap and
camel-cxf-spring-rest suites pass too.
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
