[camel] branch camel-3.20.x updated: CAMEL-19738: camel-core - Loop EIP can have wrong pending inflight in case of early loop exit due to exception

[davsclaus]

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

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


The following commit(s) were added to refs/heads/camel-3.20.x by this push:
     new b7c39463c31 CAMEL-19738: camel-core - Loop EIP can have wrong pending 
inflight in case of early loop exit due to exception
b7c39463c31 is described below

commit b7c39463c31a6f97e121cdcad289b8ad5da150f9
Author: Claus Ibsen <claus.ib...@gmail.com>
AuthorDate: Sat Aug 12 18:30:30 2023 +0200

    CAMEL-19738: camel-core - Loop EIP can have wrong pending inflight in case 
of early loop exit due to exception
---
 .../java/org/apache/camel/processor/LoopProcessor.java     | 14 ++++++++++++++
 .../java/org/apache/camel/processor/LoopExceptionTest.java |  1 +
 2 files changed, 15 insertions(+)

diff --git 
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/LoopProcessor.java
 
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/LoopProcessor.java
index f4709f8dfdc..33f9b7f4930 100644
--- 
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/LoopProcessor.java
+++ 
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/LoopProcessor.java
@@ -163,12 +163,26 @@ public class LoopProcessor extends DelegateAsyncProcessor 
implements Traceable,
                     if (LOG.isTraceEnabled()) {
                         LOG.trace("Processing complete for exchangeId: {} >>> 
{}", exchange.getExchangeId(), exchange);
                     }
+                    if (!cont && expression != null) {
+                        // if we should stop due to an exception etc, then 
make sure to dec task count
+                        int gap = count - index;
+                        while (gap-- > 0) {
+                            taskCount.decrement();
+                        }
+                    }
                     callback.done(false);
                 }
             } catch (Exception e) {
                 if (LOG.isTraceEnabled()) {
                     LOG.trace("Processing failed for exchangeId: {} >>> {}", 
exchange.getExchangeId(), e.getMessage());
                 }
+                if (expression != null) {
+                    // if we should stop due to an exception etc, then make 
sure to dec task count
+                    int gap = count - index;
+                    while (gap-- > 0) {
+                        taskCount.decrement();
+                    }
+                }
                 exchange.setException(e);
                 callback.done(false);
             }
diff --git 
a/core/camel-core/src/test/java/org/apache/camel/processor/LoopExceptionTest.java
 
b/core/camel-core/src/test/java/org/apache/camel/processor/LoopExceptionTest.java
index 976ad39f8e8..1c7795cadc6 100644
--- 
a/core/camel-core/src/test/java/org/apache/camel/processor/LoopExceptionTest.java
+++ 
b/core/camel-core/src/test/java/org/apache/camel/processor/LoopExceptionTest.java
@@ -26,6 +26,7 @@ public class LoopExceptionTest extends ContextTestSupport {
     public void testLoopException() throws Exception {
         getMockEndpoint("mock:dead").expectedMessageCount(1);
         getMockEndpoint("mock:loop").expectedMessageCount(1);
+        getMockEndpoint("mock:result").expectedMessageCount(0);
 
         template.sendBody("direct:start", "Hello World");