camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/2] camel git commit: CAMEL-10272: Provide an option to stop further processing when an exception is thrown from an aggregation strategy while parallelProcessing is used.
Date Sun, 01 Jan 2017 15:00:39 GMT
Repository: camel
Updated Branches:
  refs/heads/master 7ee49fe05 -> 3a227f2c4


CAMEL-10272: Provide an option to stop further processing when an exception is thrown from
an aggregation strategy while parallelProcessing is used.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/43036a4a
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/43036a4a
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/43036a4a

Branch: refs/heads/master
Commit: 43036a4a574873feaa421abbb5393f865714d790
Parents: 7ee49fe
Author: aldettinger <aldettinger@gmail.com>
Authored: Tue Dec 27 10:55:18 2016 +0100
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Sun Jan 1 15:21:10 2017 +0100

----------------------------------------------------------------------
 .../java/org/apache/camel/RecipientList.java    |  1 +
 .../apache/camel/component/bean/MethodInfo.java |  1 +
 .../apache/camel/model/MulticastDefinition.java | 27 +++++++-
 .../camel/model/RecipientListDefinition.java    | 26 ++++++++
 .../org/apache/camel/model/SplitDefinition.java | 27 +++++++-
 .../camel/processor/MulticastProcessor.java     | 27 ++++++--
 .../apache/camel/processor/RecipientList.java   | 12 +++-
 .../camel/processor/RecipientListProcessor.java | 11 +++-
 .../org/apache/camel/processor/Splitter.java    | 18 ++++--
 .../aggregate/AggregationStrategy.java          |  2 +-
 ...ggregationStrategyThrowingExceptionTest.java | 68 ++++++++++++++++++++
 ...ggregationStrategyThrowingExceptionTest.java | 66 +++++++++++++++++++
 ...ggregationStrategyThrowingExceptionTest.java | 66 +++++++++++++++++++
 .../camel/management/ManagedMulticastTest.java  |  2 +-
 .../management/ManagedRecipientListTest.java    |  2 +-
 .../camel/management/ManagedSplitterTest.java   |  2 +-
 16 files changed, 338 insertions(+), 20 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/RecipientList.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/RecipientList.java b/camel-core/src/main/java/org/apache/camel/RecipientList.java
index 7cd9cda..bd5e996 100644
--- a/camel-core/src/main/java/org/apache/camel/RecipientList.java
+++ b/camel-core/src/main/java/org/apache/camel/RecipientList.java
@@ -48,6 +48,7 @@ public @interface RecipientList {
     boolean parallelProcessing() default false;
     boolean parallelAggregate() default false;
     boolean stopOnException() default false;
+    boolean stopOnAggregateException() default false;
     boolean streaming() default false;
     boolean ignoreInvalidEndpoints() default false;
     String strategyRef() default "";

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java b/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
index 5ee4b46..3e5a314 100644
--- a/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
+++ b/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
@@ -163,6 +163,7 @@ public class MethodInfo {
                 && matchContext(recipientListAnnotation.context())) {
             recipientList = new RecipientList(camelContext, recipientListAnnotation.delimiter());
             recipientList.setStopOnException(recipientListAnnotation.stopOnException());
+            recipientList.setStopOnAggregateException(recipientListAnnotation.stopOnAggregateException());
             recipientList.setIgnoreInvalidEndpoints(recipientListAnnotation.ignoreInvalidEndpoints());
             recipientList.setParallelProcessing(recipientListAnnotation.parallelProcessing());
             recipientList.setParallelAggregate(recipientListAnnotation.parallelAggregate());

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java b/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java
index bc8e76c..7bff217 100644
--- a/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/MulticastDefinition.java
@@ -73,6 +73,8 @@ public class MulticastDefinition extends OutputDefinition<MulticastDefinition>
i
     private Boolean shareUnitOfWork;
     @XmlAttribute
     private Boolean parallelAggregate;
+    @XmlAttribute
+    private Boolean stopOnAggregateException;
 
     public MulticastDefinition() {
     }
@@ -183,6 +185,20 @@ public class MulticastDefinition extends OutputDefinition<MulticastDefinition>
i
         setParallelAggregate(true);
         return this;
     }
+    
+    /**
+     * If enabled, unwind exceptions occurring at aggregation time to the error handler when
parallelProcessing is used.
+     * Currently, aggregation time exceptions do not stop the route processing when parallelProcessing
is used.
+     * Enabling this option allows to work around this behavior.
+     *
+     * The default value is <code>false</code> for the sake of backward compatibility.
+     *
+     * @return the builder
+     */
+    public MulticastDefinition stopOnAggregateException() {
+        setStopOnAggregateException(true);
+        return this;
+    }
 
     /**
      * If enabled then Camel will process replies out-of-order, eg in the order they come
back.
@@ -294,6 +310,7 @@ public class MulticastDefinition extends OutputDefinition<MulticastDefinition>
i
         boolean isStreaming = getStreaming() != null && getStreaming();
         boolean isStopOnException = getStopOnException() != null && getStopOnException();
         boolean isParallelAggregate = getParallelAggregate() != null && getParallelAggregate();
+        boolean isStopOnAggregateException = getStopOnAggregateException() != null &&
getStopOnAggregateException();
 
         boolean shutdownThreadPool = ProcessorDefinitionHelper.willCreateNewThreadPool(routeContext,
this, isParallelProcessing);
         ExecutorService threadPool = ProcessorDefinitionHelper.getConfiguredExecutorService(routeContext,
"Multicast", this, isParallelProcessing);
@@ -307,7 +324,7 @@ public class MulticastDefinition extends OutputDefinition<MulticastDefinition>
i
         }
 
         MulticastProcessor answer = new MulticastProcessor(routeContext.getCamelContext(),
list, strategy, isParallelProcessing,
-                                      threadPool, shutdownThreadPool, isStreaming, isStopOnException,
timeout, onPrepare, isShareUnitOfWork, isParallelAggregate);
+                                      threadPool, shutdownThreadPool, isStreaming, isStopOnException,
timeout, onPrepare, isShareUnitOfWork, isParallelAggregate, isStopOnAggregateException);
         return answer;
     }
 
@@ -474,4 +491,12 @@ public class MulticastDefinition extends OutputDefinition<MulticastDefinition>
i
         this.parallelAggregate = parallelAggregate;
     }
 
+    public Boolean getStopOnAggregateException() {
+        return stopOnAggregateException;
+    }
+
+    public void setStopOnAggregateException(Boolean stopOnAggregateException) {
+        this.stopOnAggregateException = stopOnAggregateException;
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
b/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
index 0d02a48..b7b3b85 100644
--- a/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/RecipientListDefinition.java
@@ -83,6 +83,8 @@ public class RecipientListDefinition<Type extends ProcessorDefinition<Type>>
ext
     private Integer cacheSize;
     @XmlAttribute
     private Boolean parallelAggregate;
+    @XmlAttribute
+    private Boolean stopOnAggregateException;
 
     public RecipientListDefinition() {
     }
@@ -115,6 +117,7 @@ public class RecipientListDefinition<Type extends ProcessorDefinition<Type>>
ext
         boolean isShareUnitOfWork = getShareUnitOfWork() != null && getShareUnitOfWork();
         boolean isStopOnException = getStopOnException() != null && getStopOnException();
         boolean isIgnoreInvalidEndpoints = getIgnoreInvalidEndpoints() != null &&
getIgnoreInvalidEndpoints();
+        boolean isStopOnAggregateException = getStopOnAggregateException() != null &&
getStopOnAggregateException();
 
         RecipientList answer;
         if (delimiter != null) {
@@ -129,6 +132,7 @@ public class RecipientListDefinition<Type extends ProcessorDefinition<Type>>
ext
         answer.setShareUnitOfWork(isShareUnitOfWork);
         answer.setStopOnException(isStopOnException);
         answer.setIgnoreInvalidEndpoints(isIgnoreInvalidEndpoints);
+        answer.setStopOnAggregateException(isStopOnAggregateException);
         if (getCacheSize() != null) {
             answer.setCacheSize(getCacheSize());
         }
@@ -322,6 +326,20 @@ public class RecipientListDefinition<Type extends ProcessorDefinition<Type>>
ext
     }
 
     /**
+     * If enabled, unwind exceptions occurring at aggregation time to the error handler when
parallelProcessing is used.
+     * Currently, aggregation time exceptions do not stop the route processing when parallelProcessing
is used.
+     * Enabling this option allows to work around this behavior.
+     *
+     * The default value is <code>false</code> for the sake of backward compatibility.
+     *
+     * @return the builder
+     */
+    public RecipientListDefinition<Type> stopOnAggregateException() {
+        setStopOnAggregateException(true);
+        return this;
+    }
+
+    /**
      * If enabled then Camel will process replies out-of-order, eg in the order they come
back.
      * If disabled, Camel will process replies in the same order as defined by the recipient
list.
      *
@@ -599,4 +617,12 @@ public class RecipientListDefinition<Type extends ProcessorDefinition<Type>>
ext
     public void setParallelAggregate(Boolean parallelAggregate) {
         this.parallelAggregate = parallelAggregate;
     }
+
+    public Boolean getStopOnAggregateException() {
+        return stopOnAggregateException;
+    }
+
+    public void setStopOnAggregateException(Boolean stopOnAggregateException) {
+        this.stopOnAggregateException = stopOnAggregateException;
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java b/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
index f98780f..e7305e8 100644
--- a/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/SplitDefinition.java
@@ -72,6 +72,8 @@ public class SplitDefinition extends ExpressionNode implements ExecutorServiceAw
     private Boolean shareUnitOfWork;
     @XmlAttribute
     private Boolean parallelAggregate;
+    @XmlAttribute
+    private Boolean stopOnAggregateException;
 
     public SplitDefinition() {
     }
@@ -103,6 +105,7 @@ public class SplitDefinition extends ExpressionNode implements ExecutorServiceAw
         boolean isStreaming = getStreaming() != null && getStreaming();
         boolean isShareUnitOfWork = getShareUnitOfWork() != null && getShareUnitOfWork();
         boolean isParallelAggregate = getParallelAggregate() != null && getParallelAggregate();
+        boolean isStopOnAggregateException = getStopOnAggregateException() != null &&
getStopOnAggregateException();
         boolean shutdownThreadPool = ProcessorDefinitionHelper.willCreateNewThreadPool(routeContext,
this, isParallelProcessing);
         ExecutorService threadPool = ProcessorDefinitionHelper.getConfiguredExecutorService(routeContext,
"Split", this, isParallelProcessing);
 
@@ -118,7 +121,7 @@ public class SplitDefinition extends ExpressionNode implements ExecutorServiceAw
 
         Splitter answer = new Splitter(routeContext.getCamelContext(), exp, childProcessor,
aggregationStrategy,
                             isParallelProcessing, threadPool, shutdownThreadPool, isStreaming,
isStopOnException(),
-                            timeout, onPrepare, isShareUnitOfWork, isParallelAggregate);
+                            timeout, onPrepare, isShareUnitOfWork, isParallelAggregate, isStopOnAggregateException);
         return answer;
     }
 
@@ -231,6 +234,20 @@ public class SplitDefinition extends ExpressionNode implements ExecutorServiceAw
         setParallelAggregate(true);
         return this;
     }
+    
+    /**
+     * If enabled, unwind exceptions occurring at aggregation time to the error handler when
parallelProcessing is used.
+     * Currently, aggregation time exceptions do not stop the route processing when parallelProcessing
is used.
+     * Enabling this option allows to work around this behavior.
+     *
+     * The default value is <code>false</code> for the sake of backward compatibility.
+     *
+     * @return the builder
+     */
+    public SplitDefinition stopOnAggregateException() {
+        setStopOnAggregateException(true);
+        return this;
+    }
 
     /**
      * When in streaming mode, then the splitter splits the original message on-demand, and
each splitted
@@ -390,6 +407,14 @@ public class SplitDefinition extends ExpressionNode implements ExecutorServiceAw
     public void setParallelAggregate(Boolean parallelAggregate) {
         this.parallelAggregate = parallelAggregate;
     }
+    
+    public Boolean getStopOnAggregateException() {
+        return this.stopOnAggregateException;
+    }
+
+    public void setStopOnAggregateException(Boolean stopOnAggregateException) {
+        this.stopOnAggregateException = stopOnAggregateException;
+    }
 
     public Boolean getStopOnException() {
         return stopOnException;

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
index e0cd13d..b0def97 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
@@ -152,6 +152,7 @@ public class MulticastProcessor extends ServiceSupport implements AsyncProcessor
     private final boolean parallelProcessing;
     private final boolean streaming;
     private final boolean parallelAggregate;
+    private final boolean stopOnAggregateException;
     private final boolean stopOnException;
     private final ExecutorService executorService;
     private final boolean shutdownExecutorService;
@@ -176,10 +177,17 @@ public class MulticastProcessor extends ServiceSupport implements AsyncProcessor
                 streaming, stopOnException, timeout, onPrepare, shareUnitOfWork, false);
     }
 
+    public MulticastProcessor(CamelContext camelContext, Collection<Processor> processors,
AggregationStrategy aggregationStrategy, boolean parallelProcessing,
+                              ExecutorService executorService, boolean shutdownExecutorService,
boolean streaming, boolean stopOnException, long timeout, Processor onPrepare,
+                              boolean shareUnitOfWork, boolean parallelAggregate) {
+        this(camelContext, processors, aggregationStrategy, parallelProcessing, executorService,
shutdownExecutorService, streaming, stopOnException, timeout, onPrepare,
+             shareUnitOfWork, false, false);
+    }
+    
     public MulticastProcessor(CamelContext camelContext, Collection<Processor> processors,
AggregationStrategy aggregationStrategy,
                               boolean parallelProcessing, ExecutorService executorService,
boolean shutdownExecutorService, boolean streaming,
                               boolean stopOnException, long timeout, Processor onPrepare,
boolean shareUnitOfWork,
-                              boolean parallelAggregate) {
+                              boolean parallelAggregate, boolean stopOnAggregateException)
{
         notNull(camelContext, "camelContext");
         this.camelContext = camelContext;
         this.processors = processors;
@@ -194,6 +202,7 @@ public class MulticastProcessor extends ServiceSupport implements AsyncProcessor
         this.onPrepare = onPrepare;
         this.shareUnitOfWork = shareUnitOfWork;
         this.parallelAggregate = parallelAggregate;
+        this.stopOnAggregateException = stopOnAggregateException;
     }
 
     @Override
@@ -530,10 +539,14 @@ public class MulticastProcessor extends ServiceSupport implements AsyncProcessor
                     doAggregate(getAggregationStrategy(subExchange), result, subExchange);
                 }
             } catch (Throwable e) {
-                // wrap in exception to explain where it failed
-                CamelExchangeException cex = new CamelExchangeException("Parallel processing
failed for number " + aggregated.get(), subExchange, e);
-                subExchange.setException(cex);
-                LOG.debug(cex.getMessage(), cex);
+                if (isStopOnAggregateException()) {
+                    throw e;
+                } else {
+                    // wrap in exception to explain where it failed
+                    CamelExchangeException cex = new CamelExchangeException("Parallel processing
failed for number " + aggregated.get(), subExchange, e);
+                    subExchange.setException(cex);
+                    LOG.debug(cex.getMessage(), cex);
+                }
             } finally {
                 aggregated.incrementAndGet();
             }
@@ -1294,6 +1307,10 @@ public class MulticastProcessor extends ServiceSupport implements AsyncProcessor
         return parallelAggregate;
     }
 
+    public boolean isStopOnAggregateException() {
+        return stopOnAggregateException;
+    }
+
     public boolean isShareUnitOfWork() {
         return shareUnitOfWork;
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java b/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java
index ded8ca9..7534e87 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/RecipientList.java
@@ -61,6 +61,7 @@ public class RecipientList extends ServiceSupport implements AsyncProcessor,
IdA
     private final String delimiter;
     private boolean parallelProcessing;
     private boolean parallelAggregate;
+    private boolean stopOnAggregateException;
     private boolean stopOnException;
     private boolean ignoreInvalidEndpoints;
     private boolean streaming;
@@ -145,7 +146,8 @@ public class RecipientList extends ServiceSupport implements AsyncProcessor,
IdA
 
         RecipientListProcessor rlp = new RecipientListProcessor(exchange.getContext(), producerCache,
iter, getAggregationStrategy(),
                 isParallelProcessing(), getExecutorService(), isShutdownExecutorService(),
-                isStreaming(), isStopOnException(), getTimeout(), getOnPrepare(), isShareUnitOfWork(),
isParallelAggregate()) {
+                isStreaming(), isStopOnException(), getTimeout(), getOnPrepare(), isShareUnitOfWork(),
isParallelAggregate(),
+                isStopOnAggregateException()) {
             @Override
             protected synchronized ExecutorService createAggregateExecutorService(String
name) {
                 // use a shared executor service to avoid creating new thread pools
@@ -250,6 +252,14 @@ public class RecipientList extends ServiceSupport implements AsyncProcessor,
IdA
         this.parallelAggregate = parallelAggregate;
     }
 
+    public boolean isStopOnAggregateException() {
+        return stopOnAggregateException;
+    }
+
+    public void setStopOnAggregateException(boolean stopOnAggregateException) {
+        this.stopOnAggregateException = stopOnAggregateException;
+    }
+
     public boolean isStopOnException() {
         return stopOnException;
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
b/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
index db6af86..33ef611 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
@@ -163,8 +163,15 @@ public class RecipientListProcessor extends MulticastProcessor {
     public RecipientListProcessor(CamelContext camelContext, ProducerCache producerCache,
Iterator<Object> iter, AggregationStrategy aggregationStrategy,
                                   boolean parallelProcessing, ExecutorService executorService,
boolean shutdownExecutorService,
                                   boolean streaming, boolean stopOnException, long timeout,
Processor onPrepare, boolean shareUnitOfWork, boolean parallelAggregate) {
-        super(camelContext, null, aggregationStrategy, parallelProcessing, executorService,
shutdownExecutorService,
-                streaming, stopOnException, timeout, onPrepare, shareUnitOfWork, parallelAggregate);
+        this(camelContext, producerCache, iter, aggregationStrategy, parallelProcessing,
executorService, shutdownExecutorService, streaming, stopOnException, timeout, onPrepare,
+             shareUnitOfWork, parallelAggregate, false);
+    }
+
+    public RecipientListProcessor(CamelContext camelContext, ProducerCache producerCache,
Iterator<Object> iter, AggregationStrategy aggregationStrategy,
+                                  boolean parallelProcessing, ExecutorService executorService,
boolean shutdownExecutorService, boolean streaming, boolean stopOnException,
+                                  long timeout, Processor onPrepare, boolean shareUnitOfWork,
boolean parallelAggregate, boolean stopOnAggregateException) {
+        super(camelContext, null, aggregationStrategy, parallelProcessing, executorService,
shutdownExecutorService, streaming, stopOnException, timeout, onPrepare,
+              shareUnitOfWork, parallelAggregate, stopOnAggregateException);
         this.producerCache = producerCache;
         this.iter = iter;
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/Splitter.java b/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
index ba3be2e..8a06f79 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
@@ -69,12 +69,18 @@ public class Splitter extends MulticastProcessor implements AsyncProcessor,
Trac
                 streaming, stopOnException, timeout, onPrepare, useSubUnitOfWork, false);
     }
 
-    public Splitter(CamelContext camelContext, Expression expression, Processor destination,
AggregationStrategy aggregationStrategy,
-                    boolean parallelProcessing, ExecutorService executorService, boolean
shutdownExecutorService,
-                    boolean streaming, boolean stopOnException, long timeout, Processor onPrepare,
boolean useSubUnitOfWork,
-                    boolean parallelAggregate) {
-        super(camelContext, Collections.singleton(destination), aggregationStrategy, parallelProcessing,
executorService,
-                shutdownExecutorService, streaming, stopOnException, timeout, onPrepare,
useSubUnitOfWork, parallelAggregate);
+    public Splitter(CamelContext camelContext, Expression expression, Processor destination,
AggregationStrategy aggregationStrategy, boolean parallelProcessing,
+                    ExecutorService executorService, boolean shutdownExecutorService, boolean
streaming, boolean stopOnException, long timeout, Processor onPrepare,
+                    boolean useSubUnitOfWork, boolean parallelAggregate) {
+        this(camelContext, expression, destination, aggregationStrategy, parallelProcessing,
executorService, shutdownExecutorService, streaming, stopOnException, timeout,
+             onPrepare, useSubUnitOfWork, false, false);
+    }
+
+    public Splitter(CamelContext camelContext, Expression expression, Processor destination,
AggregationStrategy aggregationStrategy, boolean parallelProcessing,
+                    ExecutorService executorService, boolean shutdownExecutorService, boolean
streaming, boolean stopOnException, long timeout, Processor onPrepare,
+                    boolean useSubUnitOfWork, boolean parallelAggregate, boolean stopOnAggregateException)
{
+        super(camelContext, Collections.singleton(destination), aggregationStrategy, parallelProcessing,
executorService, shutdownExecutorService, streaming, stopOnException,
+              timeout, onPrepare, useSubUnitOfWork, parallelAggregate, stopOnAggregateException);
         this.expression = expression;
         notNull(expression, "expression");
         notNull(destination, "destination");

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationStrategy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationStrategy.java
b/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationStrategy.java
index 802e1b8..c593fa4 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationStrategy.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregationStrategy.java
@@ -29,7 +29,7 @@ import org.apache.camel.Exchange;
  * could be to sum up a total amount etc.
  * <p/>
  * Note that <tt>oldExchange</tt> may be <tt>null</tt> more than
once when this strategy is throwing a {@link java.lang.RuntimeException}
- * and <tt>parallelProcessing</tt> is used.
+ * and <tt>parallelProcessing</tt> is used. You can work around this behavior
using the <tt>stopOnAggregateException</tt> option.
  * <p/>
  * It is possible that <tt>newExchange</tt> is <tt>null</tt> which
could happen if there was no data possible
  * to acquire. Such as when using a {@link org.apache.camel.processor.PollEnricher} to poll
from a JMS queue which

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/test/java/org/apache/camel/issues/MulticastParallelWithAggregationStrategyThrowingExceptionTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/issues/MulticastParallelWithAggregationStrategyThrowingExceptionTest.java
b/camel-core/src/test/java/org/apache/camel/issues/MulticastParallelWithAggregationStrategyThrowingExceptionTest.java
new file mode 100644
index 0000000..4b35ea8
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/issues/MulticastParallelWithAggregationStrategyThrowingExceptionTest.java
@@ -0,0 +1,68 @@
+/**
+ * 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.issues;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+
+/**
+ * Tests the issue stated in
+ * <a href="https://issues.apache.org/jira/browse/CAMEL-10272">CAMEL-10272</a>.
+ */
+public class MulticastParallelWithAggregationStrategyThrowingExceptionTest extends ContextTestSupport
{
+
+    public void testAggregationTimeExceptionWithParallelProcessing() throws Exception {
+        getMockEndpoint("mock:a").expectedMessageCount(1);
+        getMockEndpoint("mock:b").expectedMessageCount(1);
+        getMockEndpoint("mock:end").expectedMessageCount(0);
+        getMockEndpoint("mock:dead").expectedMessageCount(1);
+
+        template.sendBody("direct:start", "Hello World");
+
+        assertMockEndpointsSatisfied();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                errorHandler(deadLetterChannel("mock:dead"));
+
+                // must use share UoW if we want the error handler to react on
+                // exceptions
+                // from the aggregation strategy also.
+                from("direct:start").
+                multicast(new MyAggregateBean()).parallelProcessing().stopOnAggregateException().shareUnitOfWork()
+                    .to("mock:a")
+                    .to("mock:b")
+               .end()
+                    .to("mock:end");
+            }
+        };
+    }
+
+    public static class MyAggregateBean implements AggregationStrategy {
+
+        public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+            throw new RuntimeException("Simulating a runtime exception thrown from the aggregation
strategy");
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/test/java/org/apache/camel/issues/RecipientListParallelWithAggregationStrategyThrowingExceptionTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/issues/RecipientListParallelWithAggregationStrategyThrowingExceptionTest.java
b/camel-core/src/test/java/org/apache/camel/issues/RecipientListParallelWithAggregationStrategyThrowingExceptionTest.java
new file mode 100644
index 0000000..4509c23
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/issues/RecipientListParallelWithAggregationStrategyThrowingExceptionTest.java
@@ -0,0 +1,66 @@
+/**
+ * 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.issues;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+
+/**
+ * Tests the issue stated in
+ * <a href="https://issues.apache.org/jira/browse/CAMEL-10272">CAMEL-10272</a>.
+ */
+public class RecipientListParallelWithAggregationStrategyThrowingExceptionTest extends ContextTestSupport
{
+
+    public void testAggregationTimeExceptionWithParallelProcessing() throws Exception {
+        getMockEndpoint("mock:a").expectedMessageCount(1);
+        getMockEndpoint("mock:b").expectedMessageCount(1);
+        getMockEndpoint("mock:end").expectedMessageCount(0);
+        getMockEndpoint("mock:dead").expectedMessageCount(1);
+
+        template.sendBodyAndHeader("direct:start", "Hello World", "recipients", "mock:a,mock:b");
+
+        assertMockEndpointsSatisfied();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                errorHandler(deadLetterChannel("mock:dead"));
+
+                // must use share UoW if we want the error handler to react on
+                // exceptions
+                // from the aggregation strategy also.
+                from("direct:start").
+                recipientList(header("recipients")).aggregationStrategy(new MyAggregateBean()).
+                parallelProcessing().stopOnAggregateException().shareUnitOfWork()
+                .end()
+                    .to("mock:end");
+            }
+        };
+    }
+
+    public static class MyAggregateBean implements AggregationStrategy {
+
+        public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+            throw new RuntimeException("Simulating a runtime exception thrown from the aggregation
strategy");
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelWithAggregationStrategyThrowingExceptionTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelWithAggregationStrategyThrowingExceptionTest.java
b/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelWithAggregationStrategyThrowingExceptionTest.java
new file mode 100644
index 0000000..66743f4
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelWithAggregationStrategyThrowingExceptionTest.java
@@ -0,0 +1,66 @@
+/**
+ * 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.issues;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+
+/**
+ * Tests the issue stated in
+ * <a href="https://issues.apache.org/jira/browse/CAMEL-10272">CAMEL-10272</a>.
+ */
+public class SplitterParallelWithAggregationStrategyThrowingExceptionTest extends ContextTestSupport
{
+
+    public void testAggregationTimeExceptionWithParallelProcessing() throws Exception {
+        getMockEndpoint("mock:a").expectedMessageCount(2);
+        getMockEndpoint("mock:end").expectedMessageCount(0);
+        getMockEndpoint("mock:dead").expectedMessageCount(1);
+
+        template.sendBody("direct:start", "Hello@World");
+
+        assertMockEndpointsSatisfied();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                errorHandler(deadLetterChannel("mock:dead"));
+
+                // must use share UoW if we want the error handler to react on
+                // exceptions
+                // from the aggregation strategy also.
+                from("direct:start").
+                split(body().tokenize("@")).aggregationStrategy(new MyAggregateBean()).
+                parallelProcessing().stopOnAggregateException().shareUnitOfWork()
+                    .to("mock:a")
+               .end()
+                    .to("mock:end");
+            }
+        };
+    }
+
+    public static class MyAggregateBean implements AggregationStrategy {
+
+        public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+            throw new RuntimeException("Simulating a runtime exception thrown from the aggregation
strategy");
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/test/java/org/apache/camel/management/ManagedMulticastTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedMulticastTest.java
b/camel-core/src/test/java/org/apache/camel/management/ManagedMulticastTest.java
index 8adac73..486bccf 100644
--- a/camel-core/src/test/java/org/apache/camel/management/ManagedMulticastTest.java
+++ b/camel-core/src/test/java/org/apache/camel/management/ManagedMulticastTest.java
@@ -67,7 +67,7 @@ public class ManagedMulticastTest extends ManagementTestSupport {
 
         data = (TabularData) mbeanServer.invoke(name, "explain", new Object[]{true}, new
String[]{"boolean"});
         assertNotNull(data);
-        assertEquals(14, data.size());
+        assertEquals(15, data.size());
 
         String json = (String) mbeanServer.invoke(name, "informationJson", null, null);
         assertNotNull(json);

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/test/java/org/apache/camel/management/ManagedRecipientListTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedRecipientListTest.java
b/camel-core/src/test/java/org/apache/camel/management/ManagedRecipientListTest.java
index 0e041e3..a37d6ca 100644
--- a/camel-core/src/test/java/org/apache/camel/management/ManagedRecipientListTest.java
+++ b/camel-core/src/test/java/org/apache/camel/management/ManagedRecipientListTest.java
@@ -91,7 +91,7 @@ public class ManagedRecipientListTest extends ManagementTestSupport {
 
         data = (TabularData) mbeanServer.invoke(on, "explain", new Object[]{true}, new String[]{"boolean"});
         assertNotNull(data);
-        assertEquals(17, data.size());
+        assertEquals(18, data.size());
 
         String json = (String) mbeanServer.invoke(on, "informationJson", null, null);
         assertNotNull(json);

http://git-wip-us.apache.org/repos/asf/camel/blob/43036a4a/camel-core/src/test/java/org/apache/camel/management/ManagedSplitterTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedSplitterTest.java
b/camel-core/src/test/java/org/apache/camel/management/ManagedSplitterTest.java
index c13b342..3276f9e 100644
--- a/camel-core/src/test/java/org/apache/camel/management/ManagedSplitterTest.java
+++ b/camel-core/src/test/java/org/apache/camel/management/ManagedSplitterTest.java
@@ -75,7 +75,7 @@ public class ManagedSplitterTest extends ManagementTestSupport {
 
         data = (TabularData) mbeanServer.invoke(on, "explain", new Object[]{true}, new String[]{"boolean"});
         assertNotNull(data);
-        assertEquals(15, data.size());
+        assertEquals(16, data.size());
 
         String json = (String) mbeanServer.invoke(on, "informationJson", null, null);
         assertNotNull(json);


Mime
View raw message