This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-25297-backport-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 8f0177489d7b960031aadfa57b0ef181e3d15643 Author: allthingssecurity <[email protected]> AuthorDate: Mon Oct 5 14:10:28 2026 +0530 CAMEL-25297: camel-direct, camel-kamelet - do not send to a suspended or stopped route while another exchange waits for its consumer (#27338) DirectProducer cached the consumer in two fields, stateCounter and consumer, and on a change it set stateCounter first and then looked up the consumer (which waits for the consumer with block=true, the default). While an exchange waited in that lookup, every other exchange sent with the same producer saw the new stateCounter and the old consumer, and was processed by the route that was just suspended or stopped, instead of waiting for its consumer. KameletProducer has the same code. Both producers now keep the consumer and the state counter read before the lookup together in one immutable holder, so another exchange either sees the old counter (and looks up the consumer too) or the result of the lookup. Found with a TLA+ model of the consumer cache (invariant: an exchange whose check starts after the consumer was removed is not sent to that consumer). Co-Authored-By: Claude Opus 5.5 <[email protected]> (cherry picked from commit 730e8715eb67d2fefe4cbca8527817fce75b2e36) Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/component/direct/DirectProducer.java | 33 ++++-- .../camel/component/kamelet/KameletProducer.java | 28 ++++- .../KameletProducerSuspendedConsumerTest.java | 123 +++++++++++++++++++++ .../DirectProducerSuspendedConsumerTest.java | 119 ++++++++++++++++++++ 4 files changed, 287 insertions(+), 16 deletions(-) diff --git a/components/camel-direct/src/main/java/org/apache/camel/component/direct/DirectProducer.java b/components/camel-direct/src/main/java/org/apache/camel/component/direct/DirectProducer.java index ecd3a9cf70aa..3ff25eeacf7c 100644 --- a/components/camel-direct/src/main/java/org/apache/camel/component/direct/DirectProducer.java +++ b/components/camel-direct/src/main/java/org/apache/camel/component/direct/DirectProducer.java @@ -30,8 +30,9 @@ public class DirectProducer extends DefaultAsyncProducer { private static final Logger LOG = LoggerFactory.getLogger(DirectProducer.class); - private volatile DirectConsumer consumer; - private int stateCounter; + // the consumer and the state counter of the component when the consumer was looked up, kept together as several + // threads can send with this producer at the same time + private volatile CachedConsumer cachedConsumer; private final DirectEndpoint endpoint; private final DirectComponent component; @@ -50,10 +51,7 @@ public class DirectProducer extends DefaultAsyncProducer { @Override public void process(Exchange exchange) throws Exception { - if (consumer == null || stateCounter != component.getStateCounter()) { - stateCounter = component.getStateCounter(); - consumer = component.getConsumer(key, block, timeout); - } + DirectConsumer consumer = getConsumer(); if (consumer == null) { if (endpoint.isFailIfNoConsumers()) { throw new DirectConsumerNotAvailableException("No consumers available on endpoint: " + endpoint, exchange); @@ -74,10 +72,7 @@ public class DirectProducer extends DefaultAsyncProducer { callback.done(true); return true; } - if (consumer == null || stateCounter != component.getStateCounter()) { - stateCounter = component.getStateCounter(); - consumer = component.getConsumer(key, block, timeout); - } + DirectConsumer consumer = getConsumer(); if (consumer == null) { if (endpoint.isFailIfNoConsumers()) { exchange.setException(new DirectConsumerNotAvailableException( @@ -116,4 +111,22 @@ public class DirectProducer extends DefaultAsyncProducer { } } + /** + * Gets the consumer, which is looked up again when it has been added or removed (such as when its route is + * suspended or stopped) since it was looked up last. + */ + private DirectConsumer getConsumer() throws InterruptedException { + CachedConsumer cached = cachedConsumer; + // read the counter before the lookup, so a change during the lookup makes the next exchange look up again + int stateCounter = component.getStateCounter(); + if (cached == null || cached.consumer() == null || cached.stateCounter() != stateCounter) { + DirectConsumer consumer = component.getConsumer(key, block, timeout); + cachedConsumer = new CachedConsumer(consumer, stateCounter); + return consumer; + } + return cached.consumer(); + } + + private record CachedConsumer(DirectConsumer consumer, int stateCounter) { + } } diff --git a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletProducer.java b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletProducer.java index 1d219d524d34..3715297bc09f 100644 --- a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletProducer.java +++ b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletProducer.java @@ -31,8 +31,9 @@ final class KameletProducer extends DefaultAsyncProducer implements RouteIdAware private static final Logger LOG = LoggerFactory.getLogger(KameletProducer.class); - private volatile KameletConsumer consumer; - private int stateCounter; + // the consumer and the state counter of the component when the consumer was looked up, kept together as several + // threads can send with this producer at the same time + private volatile CachedConsumer cachedConsumer; private final KameletEndpoint endpoint; private final KameletComponent component; @@ -56,10 +57,7 @@ final class KameletProducer extends DefaultAsyncProducer implements RouteIdAware @Override public boolean process(Exchange exchange, AsyncCallback callback) { try { - if (consumer == null || stateCounter != component.getStateCounter()) { - stateCounter = component.getStateCounter(); - consumer = component.getConsumer(key, block, timeout); - } + final KameletConsumer consumer = getConsumer(); if (consumer == null) { if (endpoint.isFailIfNoConsumers()) { exchange.setException(new KameletConsumerNotAvailableException( @@ -148,4 +146,22 @@ final class KameletProducer extends DefaultAsyncProducer implements RouteIdAware } } + /** + * Gets the consumer, which is looked up again when it has been added or removed (such as when its route is + * suspended or stopped) since it was looked up last. + */ + private KameletConsumer getConsumer() throws InterruptedException { + CachedConsumer cached = cachedConsumer; + // read the counter before the lookup, so a change during the lookup makes the next exchange look up again + int stateCounter = component.getStateCounter(); + if (cached == null || cached.consumer() == null || cached.stateCounter() != stateCounter) { + KameletConsumer consumer = component.getConsumer(key, block, timeout); + cachedConsumer = new CachedConsumer(consumer, stateCounter); + return consumer; + } + return cached.consumer(); + } + + private record CachedConsumer(KameletConsumer consumer, int stateCounter) { + } } diff --git a/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletProducerSuspendedConsumerTest.java b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletProducerSuspendedConsumerTest.java new file mode 100644 index 000000000000..bab44936cdd3 --- /dev/null +++ b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletProducerSuspendedConsumerTest.java @@ -0,0 +1,123 @@ +/* + * 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.kamelet; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.camel.CamelContext; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * While one exchange waits for the consumer of a suspended or stopped kamelet route, other exchanges sent with the same + * producer must wait too, and not be sent to the suspended or stopped consumer. + */ +@Timeout(30) +public class KameletProducerSuspendedConsumerTest extends CamelTestSupport { + + // counted down by each exchange that looks up the consumer of the kamelet route + private final AtomicReference<CountDownLatch> lookups = new AtomicReference<>(new CountDownLatch(0)); + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext context = super.createCamelContext(); + context.addComponent("kamelet", new KameletComponent() { + @Override + protected KameletConsumer getConsumer(String key, boolean block, long timeout) throws InterruptedException { + lookups.get().countDown(); + return super.getConsumer(key, block, timeout); + } + }); + return context; + } + + @Test + public void testExchangesWaitForSuspendedConsumer() throws Exception { + testExchangesWaitForConsumer(() -> context.getRouteController().suspendRoute("echo"), + () -> context.getRouteController().resumeRoute("echo")); + } + + @Test + public void testExchangesWaitForStoppedConsumer() throws Exception { + testExchangesWaitForConsumer(() -> context.getRouteController().stopRoute("echo"), + () -> context.getRouteController().startRoute("echo")); + } + + private void testExchangesWaitForConsumer(RouteAction removeConsumer, RouteAction addConsumer) throws Exception { + MockEndpoint mock = getMockEndpoint("mock:kamelet"); + mock.expectedBodiesReceived("warm"); + // the producer of the kamelet now caches the consumer of the kamelet route + template.sendBody("direct:start", "warm"); + mock.assertIsSatisfied(); + + removeConsumer.run(); + + CountDownLatch firstWaiting = new CountDownLatch(1); + lookups.set(firstWaiting); + CompletableFuture<Object> first = template.asyncSendBody("direct:start", "first"); + // the first exchange found that the consumer changed and waits for the consumer + assertThat(firstWaiting.await(10, TimeUnit.SECONDS)).isTrue(); + + CountDownLatch secondWaiting = new CountDownLatch(1); + lookups.set(secondWaiting); + CompletableFuture<Object> second = template.asyncSendBody("direct:start", "second"); + + // the second exchange must wait for the consumer too, and not reach the kamelet route while its consumer is gone + await().atMost(10, TimeUnit.SECONDS).until(() -> secondWaiting.getCount() == 0 || second.isDone()); + assertThat(mock.getReceivedCounter()).as("No exchange should reach the kamelet route while its consumer is gone") + .isEqualTo(1); + assertThat(second).as("The second exchange should wait for the consumer of the kamelet route").isNotDone(); + + mock.reset(); + mock.expectedBodiesReceivedInAnyOrder("first", "second"); + addConsumer.run(); + + first.get(10, TimeUnit.SECONDS); + second.get(10, TimeUnit.SECONDS); + mock.assertIsSatisfied(); + } + + @FunctionalInterface + private interface RouteAction { + void run() throws Exception; + } + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + routeTemplate("echo") + .from("kamelet:source") + .to("mock:kamelet"); + + from("direct:start") + .to("kamelet:echo/echo?timeout=20000"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerSuspendedConsumerTest.java b/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerSuspendedConsumerTest.java new file mode 100644 index 000000000000..b49b4c02c83e --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/direct/DirectProducerSuspendedConsumerTest.java @@ -0,0 +1,119 @@ +/* + * 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.direct; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.camel.CamelContext; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * While one exchange waits for the consumer of a suspended or stopped direct route, other exchanges sent with the same + * producer must wait too, and not be sent to the suspended or stopped consumer. + */ +@Timeout(30) +public class DirectProducerSuspendedConsumerTest extends ContextTestSupport { + + // counted down by each exchange that looks up the consumer of the suspended route + private final AtomicReference<CountDownLatch> lookups = new AtomicReference<>(new CountDownLatch(0)); + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext context = super.createCamelContext(); + context.addComponent("direct", new DirectComponent() { + @Override + protected DirectConsumer getConsumer(String key, boolean block, long timeout) throws InterruptedException { + lookups.get().countDown(); + return super.getConsumer(key, block, timeout); + } + }); + return context; + } + + @Test + public void testExchangesWaitForSuspendedConsumer() throws Exception { + testExchangesWaitForConsumer(() -> context.getRouteController().suspendRoute("b"), + () -> context.getRouteController().resumeRoute("b")); + } + + @Test + public void testExchangesWaitForStoppedConsumer() throws Exception { + testExchangesWaitForConsumer(() -> context.getRouteController().stopRoute("b"), + () -> context.getRouteController().startRoute("b")); + } + + private void testExchangesWaitForConsumer(RouteAction removeConsumer, RouteAction addConsumer) throws Exception { + MockEndpoint mock = getMockEndpoint("mock:b"); + mock.expectedBodiesReceived("warm"); + // the producer of direct:b now caches the consumer of route b + template.sendBody("direct:start", "warm"); + mock.assertIsSatisfied(); + + removeConsumer.run(); + + CountDownLatch firstWaiting = new CountDownLatch(1); + lookups.set(firstWaiting); + CompletableFuture<Object> first = template.asyncSendBody("direct:start", "first"); + // the first exchange found that the consumer changed and waits for the consumer + assertTrue(firstWaiting.await(10, TimeUnit.SECONDS)); + + CountDownLatch secondWaiting = new CountDownLatch(1); + lookups.set(secondWaiting); + CompletableFuture<Object> second = template.asyncSendBody("direct:start", "second"); + + // the second exchange must wait for the consumer too, and not reach route b while its consumer is gone + await().atMost(10, TimeUnit.SECONDS).until(() -> secondWaiting.getCount() == 0 || second.isDone()); + assertEquals(1, mock.getReceivedCounter(), "No exchange should reach route b while its consumer is gone"); + assertFalse(second.isDone(), "The second exchange should wait for the consumer of route b"); + + mock.reset(); + mock.expectedBodiesReceivedInAnyOrder("first", "second"); + addConsumer.run(); + + first.get(10, TimeUnit.SECONDS); + second.get(10, TimeUnit.SECONDS); + mock.assertIsSatisfied(); + } + + @FunctionalInterface + private interface RouteAction { + void run() throws Exception; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from("direct:start").to("direct:b?timeout=20000"); + + from("direct:b").routeId("b").to("mock:b"); + } + }; + } +}
