CAMEL-18275: Address issue of completions not being run in SEDA pipeline

core/camel-api/src/main/java/org/apache/camel/spi/SynchronizationVetoable.java

@@ -16,6 +16,8 @@
  */
 package org.apache.camel.spi;
 
+import org.apache.camel.Exchange;
+
 /**
  * A vetoable {@link org.apache.camel.spi.Synchronization}.
  * <p/>
@@ -40,4 +42,11 @@ public interface SynchronizationVetoable extends Synchronization {
      * @return <tt>true</tt> to allow handover, <tt>false</tt> to deny.
      */
     boolean allowHandover();
+
+    /**
+     * A method to perform optional housekeeping when a Synchronization is being handed over.
+     * 
+     * @param target The Exchange to which the synchronizations are being transferred.
+     */
+    void beforeHandover(Exchange target);
 }

core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java

@@ -210,14 +210,19 @@ public void handoverSynchronization(Exchange target, Predicate<Synchronization>
             Synchronization synchronization = it.next();
 
             boolean handover = true;
+            SynchronizationVetoable veto = null;
             if (synchronization instanceof SynchronizationVetoable) {
-                SynchronizationVetoable veto = (SynchronizationVetoable) synchronization;
+                veto = (SynchronizationVetoable) synchronization;
                 handover = veto.allowHandover();
             }
 
             if (handover && (filter == null || filter.test(synchronization))) {
                 log.trace("Handover synchronization {} to: {}", synchronization, target);
                 target.adapt(ExtendedExchange.class).addOnCompletion(synchronization);
+                // Allow the synchronization to do housekeeping before transfer
+                if (veto != null) {
+                    veto.beforeHandover(target);
+                }
                 // remove it if its handed over
                 it.remove();
             } else {

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

@@ -368,6 +368,26 @@ public String toString() {
                 return "onFailureOnly";
             }
         }
+
+        @Override
+        public void beforeHandover(Exchange target) {
+            // The onAfterRoute method will not be called after the handover
+            // To ensure that completions are called, remember the route IDs here.
+            // Assumption: the fromRouteId on the target Exchange is the route
+            // which owns the completion
+            LOG.debug("beforeHandover from Route {}", target.getFromRouteId());
+            final String exchangeRouteId = target.getFromRouteId();
+            if (routeScoped && exchangeRouteId != null && exchangeRouteId.equals(routeId)) {
+                List<String> routeIds = target.getProperty(ExchangePropertyKey.ON_COMPLETION_ROUTE_IDS, List.class);
+                if (routeIds == null) {
+                    routeIds = new ArrayList<>();
+                    target.setProperty(ExchangePropertyKey.ON_COMPLETION_ROUTE_IDS, routeIds);
+                }
+                if (!routeIds.contains(exchangeRouteId)) {
+                    routeIds.add(exchangeRouteId);
+                }
+            }
+        }
     }
 
     private final class OnCompletionSynchronizationBeforeConsumer extends SynchronizationAdapter implements Ordered {

core/camel-support/src/main/java/org/apache/camel/support/SynchronizationAdapter.java

@@ -63,4 +63,9 @@ public void onBeforeRoute(Route route, Exchange exchange) {
     public void onAfterRoute(Route route, Exchange exchange) {
         // noop
     }
+
+    @Override
+    public void beforeHandover(Exchange target) {
+        // noop
+    }
 }