This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24995 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 2dbb32b216227a5f0f825bcbbd883d1c4d0f0557 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Sep 24 13:47:04 2026 +0200 CAMEL-24995: camel-core - Wire Tap EIP: a graceful shutdown does not wait for the tapped exchanges queued in the thread pool The wire tap only counted a tapped exchange as pending once its task started running, so the tapped exchanges waiting in the thread pool queue were not pending, and a graceful shutdown did not wait for them. They are now counted from when they are submitted. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/processor/WireTapProcessor.java | 6 +- .../processor/WireTapPendingExchangesTest.java | 81 ++++++++++++++++++++++ 2 files changed, 86 insertions(+), 1 deletion(-) diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/WireTapProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/WireTapProcessor.java index a6b10784ec44..e7fa9e38bd22 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/WireTapProcessor.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/WireTapProcessor.java @@ -112,7 +112,6 @@ public class WireTapProcessor extends BaseProcessorSupport @Override public void run() { - taskCount.increment(); LOG.debug(">>>> (wiretap) {} {}", uri, exchange); asyncProcessor.process(exchange, callback); } @@ -210,6 +209,9 @@ public class WireTapProcessor extends BaseProcessorSupport } // send the exchange to the destination using an executor service + // count the task as pending from when it is submitted (not when it starts to run), so a graceful shutdown + // also waits for the tapped exchanges that are waiting in the thread pool queue + taskCount.increment(); try { // create task which has state used during routing Runnable task = taskFactory.acquire(target, null); @@ -217,6 +219,8 @@ public class WireTapProcessor extends BaseProcessorSupport task = ProcessorHelper.prepareMDCParallelTask(camelContext, task); executorService.submit(task); } catch (Exception e) { + // the task will not run + taskCount.decrement(); // in case the thread pool rejects or cannot submit the task then we need to catch // so camel error handler can react exchange.setException(e); diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/WireTapPendingExchangesTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/WireTapPendingExchangesTest.java new file mode 100644 index 000000000000..df24fc9981b3 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/WireTapPendingExchangesTest.java @@ -0,0 +1,81 @@ +/* + * 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.processor; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.builder.RouteBuilder; +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; + +/** + * The tapped exchanges waiting in the thread pool are pending exchanges, so a graceful shutdown waits for them. + */ +public class WireTapPendingExchangesTest extends ContextTestSupport { + + private final ExecutorService pool = Executors.newSingleThreadExecutor(); + + @Override + @AfterEach + public void tearDown() throws Exception { + super.tearDown(); + pool.shutdownNow(); + } + + @Test + public void testQueuedTapsArePending() throws Exception { + getMockEndpoint("mock:tap").expectedMessageCount(3); + + // occupy the single thread of the pool, so the tapped exchanges wait in its queue + CountDownLatch latch = new CountDownLatch(1); + pool.submit(() -> { + latch.await(10, TimeUnit.SECONDS); + return null; + }); + + template.sendBody("direct:start", "A"); + template.sendBody("direct:start", "B"); + template.sendBody("direct:start", "C"); + + WireTapProcessor tap = context.getProcessor("tap", WireTapProcessor.class); + assertEquals(3, tap.getPendingExchangesSize(), "the queued tapped exchanges should be pending"); + + latch.countDown(); + assertMockEndpointsSatisfied(); + // the task is done just after the tapped exchange has been received + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(0, tap.getPendingExchangesSize())); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start").wireTap("direct:tap").executorService(pool).id("tap").to("mock:result"); + + from("direct:tap").to("mock:tap"); + } + }; + } +}
