[camel] 02/02: CAMEL-16331: camel-core - DefaultConsumer used as PolllingConsumer EIP should not be pooled exchanges. Also fix little bug in ExchangeHelper copy which did not handover onCompletions if UoW was null on target.

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git

commit 6cdee841318bf88f7a389871e95bb1ba25a23323
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri Mar 12 14:06:08 2021 +0100

    CAMEL-16331: camel-core - DefaultConsumer used as PolllingConsumer EIP should not be pooled exchanges. Also fix little bug in ExchangeHelper copy which did not handover onCompletions if UoW was null on target.
---
 .../src/main/java/org/apache/camel/processor/PollEnricher.java   | 6 +++++-
 .../file/FileConsumePollEnrichFileUsingProcessorTest.java        | 3 +++
 .../src/main/java/org/apache/camel/support/ExchangeHelper.java   | 9 +++------
 3 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
index 1a03637..e9ea51e 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
@@ -438,7 +438,7 @@ public class PollEnricher extends AsyncProcessorSupport implements IdAware, Rout
     }
 
     @Override
-    protected void doStart() throws Exception {
+    protected void doBuild() throws Exception {
         if (consumerCache == null) {
             // create consumer cache if we use dynamic expressions for computing the endpoints to poll
             consumerCache = new DefaultConsumerCache(this, camelContext, cacheSize);
@@ -447,6 +447,10 @@ public class PollEnricher extends AsyncProcessorSupport implements IdAware, Rout
         if (aggregationStrategy instanceof CamelContextAware) {
             ((CamelContextAware) aggregationStrategy).setCamelContext(camelContext);
         }
+    }
+
+    @Override
+    protected void doStart() throws Exception {
         ServiceHelper.startService(consumerCache, aggregationStrategy);
     }
 
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumePollEnrichFileUsingProcessorTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumePollEnrichFileUsingProcessorTest.java
index 305f7b3..6c66cf1 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumePollEnrichFileUsingProcessorTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumePollEnrichFileUsingProcessorTest.java
@@ -86,6 +86,9 @@ public class FileConsumePollEnrichFileUsingProcessorTest extends ContextTestSupp
                             // otherwise do a rollback
                             throw new CamelExchangeException("Cannot find the data file " + name, exchange);
                         }
+
+                        // and remember to done the UoW
+                        data.getUnitOfWork().done(data);
                     }
                 }).to("mock:start");
 
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
index 4a3d45d..42d8408 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
@@ -308,12 +308,9 @@ public final class ExchangeHelper {
         // do not share the unit of work
         ExtendedExchange ce = (ExtendedExchange) copy;
         ce.setUnitOfWork(null);
-
-        // do not reuse the message id
-        // hand over on completion to the copy if we got any
-        UnitOfWork uow = exchange.getUnitOfWork();
-        if (handover && uow != null) {
-            uow.handoverSynchronization(copy, filter);
+        if (handover) {
+            // Need to hand over the completion for async invocation
+            exchange.adapt(ExtendedExchange.class).handoverCompletions(ce);
         }
         // set a correlation id so we can track back the original exchange
         copy.setProperty(ExchangePropertyKey.CORRELATION_ID, id);