CAMEL-25012: camel-core - OnCompletion EIP: make a graceful shutdown wait for the parallel onCompletion tasks

core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java

@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.LongAdder;
 
 import org.apache.camel.AsyncCallback;
 import org.apache.camel.CamelContext;
@@ -30,9 +31,11 @@
 import org.apache.camel.Predicate;
 import org.apache.camel.Processor;
 import org.apache.camel.Route;
+import org.apache.camel.ShutdownRunningTask;
 import org.apache.camel.Traceable;
 import org.apache.camel.spi.IdAware;
 import org.apache.camel.spi.RouteIdAware;
+import org.apache.camel.spi.ShutdownAware;
 import org.apache.camel.spi.StepIdAware;
 import org.apache.camel.spi.SynchronizationRouteAware;
 import org.apache.camel.support.ExchangeHelper;
@@ -47,7 +50,8 @@
 /**
  * Processor implementing <a href="http://camel.apache.org/oncompletion.html">onCompletion</a>.
  */
-public class OnCompletionProcessor extends BaseProcessorSupport implements Traceable, IdAware, RouteIdAware, StepIdAware {
+public class OnCompletionProcessor extends BaseProcessorSupport
+        implements Traceable, ShutdownAware, IdAware, RouteIdAware, StepIdAware {
 
     private static final Logger LOG = LoggerFactory.getLogger(OnCompletionProcessor.class);
 
@@ -64,6 +68,7 @@ public class OnCompletionProcessor extends BaseProcessorSupport implements Trace
     private final boolean useOriginalBody;
     private final boolean afterConsumer;
     private final boolean routeScoped;
+    private final LongAdder taskCount = new LongAdder();
 
     public OnCompletionProcessor(CamelContext camelContext, Processor processor, ExecutorService executorService,
                                  boolean shutdownExecutorService,
@@ -107,14 +112,34 @@ protected void doStop() throws Exception {
     protected void doShutdown() throws Exception {
         ServiceHelper.stopAndShutdownService(processor);
         if (shutdownExecutorService) {
-            getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
+            List<Runnable> dropped = getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
+            if (dropped != null && !dropped.isEmpty()) {
+                // the tasks still queued in the thread pool will never run, so they are no longer pending
+                taskCount.add(-dropped.size());
+            }
         }
     }
 
     public CamelContext getCamelContext() {
         return camelContext;
     }
 
+    @Override
+    public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
+        // not in use
+        return true;
+    }
+
+    @Override
+    public int getPendingExchangesSize() {
+        return taskCount.intValue();
+    }
+
+    @Override
+    public void prepareShutdown(boolean suspendOnly, boolean forced) {
+        // noop
+    }
+
     @Override
     public String getId() {
         return id;
@@ -162,6 +187,30 @@ public boolean process(Exchange exchange, AsyncCallback callback) {
         return true;
     }
 
+    /**
+     * Submits the onCompletion task to the thread pool (parallel processing). The task is counted as pending from when
+     * it is submitted until it is done, so a graceful shutdown waits for it.
+     */
+    @SuppressWarnings("deprecation")
+    private void submitTask(Runnable task) {
+        taskCount.increment();
+        Runnable counted = () -> {
+            try {
+                task.run();
+            } finally {
+                taskCount.decrement();
+            }
+        };
+        try {
+            // Deprecated since 4.19.0
+            executorService.submit(prepareMDCParallelTask(camelContext, counted));
+        } catch (RuntimeException e) {
+            // the task will not run
+            taskCount.decrement();
+            throw e;
+        }
+    }
+
     protected boolean isCreateCopy() {
         // we need to create a correlated copy if we run in parallel mode or is in after consumer mode (as the UoW would be done on the original exchange otherwise)
         return executorService != null || afterConsumer;
@@ -301,7 +350,6 @@ public void onAfterRoute(Route route, Exchange exchange) {
             };
         }
 
-        @SuppressWarnings("deprecation")
         @Override
         public void onComplete(final Exchange exchange) {
             if (shouldSkip(exchange, onFailureOnly)) {
@@ -316,17 +364,14 @@ public void onComplete(final Exchange exchange) {
                     LOG.debug("Processing onComplete: {}", copy);
                     doProcess(processor, copy);
                 };
-                // Deprecated since 4.19.0
-                task = prepareMDCParallelTask(camelContext, task);
-                executorService.submit(task);
+                submitTask(task);
             } else {
                 // run without thread-pool
                 LOG.debug("Processing onComplete: {}", copy);
                 doProcess(processor, copy);
             }
         }
 
-        @SuppressWarnings("deprecation")
         @Override
         public void onFailure(final Exchange exchange) {
             if (shouldSkip(exchange, onCompleteOnly)) {
@@ -349,9 +394,7 @@ public void onFailure(final Exchange exchange) {
                     // restore exception after processing
                     copy.setException(original);
                 };
-                // Deprecated since 4.19.0
-                task = prepareMDCParallelTask(camelContext, task);
-                executorService.submit(task);
+                submitTask(task);
             } else {
                 // run without thread-pool
                 LOG.debug("Processing onFailure: {}", copy);
@@ -444,7 +487,6 @@ public void onBeforeRoute(Route route, Exchange exchange) {
                     // NO-OP
                 }
 
-                @SuppressWarnings("deprecation")
                 @Override
                 public void onAfterRoute(Route route, Exchange exchange) {
                     LOG.debug("onAfterRoute from Route {}", route.getRouteId());
@@ -479,9 +521,7 @@ public void onAfterRoute(Route route, Exchange exchange) {
                             LOG.debug("Processing onAfterRoute: {}", copy);
                             doProcess(processor, copy);
                         };
-                        // Deprecated since 4.19.0
-                        task = prepareMDCParallelTask(camelContext, task);
-                        executorService.submit(task);
+                        submitTask(task);
                     } else {
                         // run without thread-pool
                         LOG.debug("Processing onAfterRoute: {}", copy);

core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionParallelProcessingForcedShutdownTest.java

@@ -0,0 +1,87 @@
+/*
+ * 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.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.builder.ThreadPoolProfileBuilder;
+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;
+
+/**
+ * When a graceful shutdown times out, the thread pool of a parallel onCompletion is shut down, which drops the
+ * onCompletion tasks still queued in it. They must no longer be counted as pending exchanges.
+ */
+class OnCompletionParallelProcessingForcedShutdownTest extends ContextTestSupport {
+
+    private final CountDownLatch release = new CountDownLatch(1);
+    private final AtomicInteger started = new AtomicInteger();
+
+    @Test
+    void testForcedShutdownDropsQueuedOnCompletion() throws Exception {
+        // the first onCompletion runs (and waits), the second is queued in the pool with one thread
+        template.sendBody("direct:start", "A");
+        template.sendBody("direct:start", "B");
+        await().atMost(10, TimeUnit.SECONDS).until(() -> started.get() == 1);
+        OnCompletionProcessor onCompletion = context.getProcessor("oc", OnCompletionProcessor.class);
+        assertEquals(2, onCompletion.getPendingExchangesSize());
+
+        // the graceful shutdown times out, and the pool is shut down: the running task is interrupted, and the queued
+        // task is dropped
+        context.getShutdownStrategy().setTimeout(1);
+        context.stop();
+        assertTrue(context.getShutdownStrategy().hasTimeoutOccurred());
+
+        await().atMost(10, TimeUnit.SECONDS)
+                .untilAsserted(() -> assertEquals(0, onCompletion.getPendingExchangesSize(),
+                        "No onCompletion task should be pending after the thread pool is shut down"));
+        assertEquals(1, started.get(), "The queued onCompletion should not have run");
+    }
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        CamelContext context = super.createCamelContext();
+        context.getExecutorServiceManager()
+                .registerThreadPoolProfile(new ThreadPoolProfileBuilder("oneThread").poolSize(1).maxPoolSize(1).build());
+        return context;
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("direct:start")
+                        .onCompletion().id("oc").parallelProcessing().executorService("oneThread")
+                        .process(e -> {
+                            started.incrementAndGet();
+                            release.await(20, TimeUnit.SECONDS);
+                        })
+                        .end()
+                        .to("mock:result");
+            }
+        };
+    }
+}

core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionParallelProcessingShutdownTest.java

@@ -0,0 +1,90 @@
+/*
+ * 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.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+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;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The onCompletion tasks running in the thread pool of a parallel onCompletion are pending exchanges, so a graceful
+ * shutdown waits for them.
+ */
+class OnCompletionParallelProcessingShutdownTest extends ContextTestSupport {
+
+    private final CountDownLatch onCompletionStarted = new CountDownLatch(1);
+    private final CountDownLatch camelStopping = new CountDownLatch(1);
+    private final AtomicInteger onCompletionDone = new AtomicInteger();
+    private final ExecutorService stopper = Executors.newSingleThreadExecutor();
+
+    @AfterEach
+    void shutdownStopper() {
+        stopper.shutdownNow();
+    }
+
+    @Test
+    void testGracefulShutdownWaitsForOnCompletion() throws Exception {
+        template.sendBody("direct:start", "Hello World");
+        assertTrue(onCompletionStarted.await(10, TimeUnit.SECONDS));
+        OnCompletionProcessor onCompletion = context.getProcessor("oc", OnCompletionProcessor.class);
+        assertEquals(1, onCompletion.getPendingExchangesSize(), "The running onCompletion task should be pending");
+
+        // the exchange is done, and Camel is stopped while its onCompletion is still running
+        Future<?> stop = stopper.submit(() -> {
+            context.stop();
+            return null;
+        });
+        await().atMost(10, TimeUnit.SECONDS).until(() -> context.isStopping() || context.isStopped());
+        camelStopping.countDown();
+        stop.get(30, TimeUnit.SECONDS);
+
+        assertEquals(1, onCompletionDone.get(), "The onCompletion should be done");
+        assertEquals(0, onCompletion.getPendingExchangesSize(), "No onCompletion task should be pending");
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("direct:start")
+                        .onCompletion().id("oc").parallelProcessing()
+                        .process(e -> {
+                            onCompletionStarted.countDown();
+                            // the onCompletion takes until Camel is stopping
+                            if (camelStopping.await(10, TimeUnit.SECONDS)) {
+                                onCompletionDone.incrementAndGet();
+                            }
+                        })
+                        .end()
+                        .to("mock:result");
+            }
+        };
+    }
+}

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc

@@ -782,6 +782,17 @@ seen, and a repository backed by a remote store keeps its connection while the r
 forget the ids, clear the repository with `IdempotentRepository.clear()` (or the `clear` JMX operation of
 the Idempotent Consumer).
 
+=== camel-core - a graceful shutdown waits for the parallel onCompletion tasks
+
+With `onCompletion().parallelProcessing()`, stopping or suspending a route (and stopping `CamelContext`) now waits
+for the onCompletion tasks that are running or queued in its thread pool, up to the shutdown timeout, just as it
+waits for the inflight exchanges. Previously the shutdown did not wait for them, and when the thread pool was shut
+down, the queued tasks were dropped and the running ones were interrupted.
+
+An onCompletion with `parallelProcessing` that synchronously stops its own route now waits for itself until the
+shutdown timeout occurs, and the route is then stopped forcibly. Stop the route asynchronously instead, for example
+from a separate thread or with the Control Bus `async=true` option.
+
 === Component deprecation
 
 ==== camel-minio