You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2022/01/03 20:01:41 UTC
[camel] branch main updated: CAMEL-17296: Deprecate stopOnAggregateException (#6625)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 740c8eb CAMEL-17296: Deprecate stopOnAggregateException (#6625)
740c8eb is described below
commit 740c8eb1d69a9ae1f2fda91e3b7150b6c483cd3f
Author: klease <38...@users.noreply.github.com>
AuthorDate: Mon Jan 3 21:00:04 2022 +0100
CAMEL-17296: Deprecate stopOnAggregateException (#6625)
In Splitter, RecipientList and Multicast, this option has no effect starting with Camel 3.x.
Deprecate the option in the related DSL definitions and remove it in the reifieris
and processors.
Modify the 3 unit tests in the issues package to remove the option but verify that
an Exception in the aggregate strategy stops the processing of the route and
invokes the error handler.
---
.../main/java/org/apache/camel/RecipientList.java | 7 +++---
.../org/apache/camel/model/multicast.json | 2 +-
.../org/apache/camel/model/recipientList.json | 2 +-
.../resources/org/apache/camel/model/split.json | 2 +-
.../apache/camel/model/MulticastDefinition.java | 16 ++++++++-----
.../camel/model/RecipientListDefinition.java | 15 ++++++++----
.../org/apache/camel/model/SplitDefinition.java | 25 ++++++++++++-------
.../DefaultAnnotationBasedProcessorFactory.java | 1 -
.../apache/camel/processor/MulticastProcessor.java | 18 +-------------
.../org/apache/camel/processor/RecipientList.java | 12 +---------
.../camel/processor/RecipientListProcessor.java | 5 ++--
.../java/org/apache/camel/processor/Splitter.java | 28 ++--------------------
.../org/apache/camel/reifier/MulticastReifier.java | 4 +---
.../apache/camel/reifier/RecipientListReifier.java | 2 --
.../org/apache/camel/reifier/SplitReifier.java | 5 ++--
...thAggregationStrategyThrowingExceptionTest.java | 8 +++++--
...thAggregationStrategyThrowingExceptionTest.java | 8 +++++--
...thAggregationStrategyThrowingExceptionTest.java | 8 +++++--
18 files changed, 70 insertions(+), 98 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/RecipientList.java b/core/camel-api/src/main/java/org/apache/camel/RecipientList.java
index 2bc3d3f..5fad560 100644
--- a/core/camel-api/src/main/java/org/apache/camel/RecipientList.java
+++ b/core/camel-api/src/main/java/org/apache/camel/RecipientList.java
@@ -77,11 +77,12 @@ public @interface RecipientList {
/**
* 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.
+ *
+ * @deprecated Since Camel 3.0 exceptions from the aggregate strategy are always propagated to the error handler,
+ * whether or not this option is enabled.
*/
+ @Deprecated
boolean stopOnAggregateException() default false;
/**
diff --git a/core/camel-core-model/src/generated/resources/org/apache/camel/model/multicast.json b/core/camel-core-model/src/generated/resources/org/apache/camel/model/multicast.json
index e3e4da5..2500ab7 100644
--- a/core/camel-core-model/src/generated/resources/org/apache/camel/model/multicast.json
+++ b/core/camel-core-model/src/generated/resources/org/apache/camel/model/multicast.json
@@ -23,7 +23,7 @@
"onPrepareRef": { "kind": "attribute", "displayName": "On Prepare Ref", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Uses the Processor when preparing the org.apache.camel.Exchange to be send. This can be used to deep-clone messages that should be send, or any custom logic needed before the exchange is send." },
"shareUnitOfWork": { "kind": "attribute", "displayName": "Share Unit Of Work", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Shares the org.apache.camel.spi.UnitOfWork with the parent and each of the sub messages. Multicast will by default not share unit of work between the parent exchange and each multicasted exchange. This means each sub exchange has its own ind [...]
"parallelAggregate": { "kind": "attribute", "displayName": "Parallel Aggregate", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If enabled then the aggregate method on AggregationStrategy can be called concurrently. Notice that this would require the implementation of AggregationStrategy to be implemented as thread-safe. By default this is false meaning that Camel [...]
- "stopOnAggregateException": { "kind": "attribute", "displayName": "Stop On Aggregate Exception", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "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. Enab [...]
+ "stopOnAggregateException": { "kind": "attribute", "displayName": "Stop On Aggregate Exception", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": true, "deprecationNote": "Has no effect in Camel 3.x", "autowired": false, "secret": false, "defaultValue": false, "description": "If enabled, unwind exceptions occurring at aggregation time to the error handler when parallelProcessing is used. The default value is false for the sake of backward compatibi [...]
"id": { "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the id of this node" },
"description": { "kind": "element", "displayName": "Description", "required": false, "type": "object", "javaType": "org.apache.camel.model.DescriptionDefinition", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the description of this node" }
}
diff --git a/core/camel-core-model/src/generated/resources/org/apache/camel/model/recipientList.json b/core/camel-core-model/src/generated/resources/org/apache/camel/model/recipientList.json
index 831cfb2..de1a355 100644
--- a/core/camel-core-model/src/generated/resources/org/apache/camel/model/recipientList.json
+++ b/core/camel-core-model/src/generated/resources/org/apache/camel/model/recipientList.json
@@ -27,7 +27,7 @@
"shareUnitOfWork": { "kind": "attribute", "displayName": "Share Unit Of Work", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Shares the org.apache.camel.spi.UnitOfWork with the parent and each of the sub messages. Recipient List will by default not share unit of work between the parent exchange and each recipient exchange. This means each sub exchange has its own [...]
"cacheSize": { "kind": "attribute", "displayName": "Cache Size", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the maximum size used by the org.apache.camel.spi.ProducerCache which is used to cache and reuse producers when using this recipient list, when uris are reused. Beware that when using dynamic endpoints then it affects how well the cache can be utilized. If each dynamic end [...]
"parallelAggregate": { "kind": "attribute", "displayName": "Parallel Aggregate", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If enabled then the aggregate method on AggregationStrategy can be called concurrently. Notice that this would require the implementation of AggregationStrategy to be implemented as thread-safe. By default this is false meaning that Camel [...]
- "stopOnAggregateException": { "kind": "attribute", "displayName": "Stop On Aggregate Exception", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "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. Enab [...]
+ "stopOnAggregateException": { "kind": "attribute", "displayName": "Stop On Aggregate Exception", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": true, "deprecationNote": "Has no effect in Camel 3.x", "autowired": false, "secret": false, "defaultValue": false, "description": "If enabled, unwind exceptions occurring at aggregation time to the error handler when parallelProcessing is used. The default value is false for the sake of backward compatibi [...]
"id": { "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the id of this node" },
"description": { "kind": "element", "displayName": "Description", "required": false, "type": "object", "javaType": "org.apache.camel.model.DescriptionDefinition", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the description of this node" }
}
diff --git a/core/camel-core-model/src/generated/resources/org/apache/camel/model/split.json b/core/camel-core-model/src/generated/resources/org/apache/camel/model/split.json
index b65ba55..361919f 100644
--- a/core/camel-core-model/src/generated/resources/org/apache/camel/model/split.json
+++ b/core/camel-core-model/src/generated/resources/org/apache/camel/model/split.json
@@ -25,7 +25,7 @@
"onPrepareRef": { "kind": "attribute", "displayName": "On Prepare Ref", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Uses the Processor when preparing the org.apache.camel.Exchange to be send. This can be used to deep-clone messages that should be send, or any custom logic needed before the exchange is send." },
"shareUnitOfWork": { "kind": "attribute", "displayName": "Share Unit Of Work", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Shares the org.apache.camel.spi.UnitOfWork with the parent and each of the sub messages. Splitter will by default not share unit of work between the parent exchange and each splitted exchange. This means each splitted exchange has its own in [...]
"parallelAggregate": { "kind": "attribute", "displayName": "Parallel Aggregate", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If enabled then the aggregate method on AggregationStrategy can be called concurrently. Notice that this would require the implementation of AggregationStrategy to be implemented as thread-safe. By default this is false meaning that Camel [...]
- "stopOnAggregateException": { "kind": "attribute", "displayName": "Stop On Aggregate Exception", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "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. Enab [...]
+ "stopOnAggregateException": { "kind": "attribute", "displayName": "Stop On Aggregate Exception", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": true, "deprecationNote": "Has no effect in Camel 3.x", "autowired": false, "secret": false, "defaultValue": false, "description": "If enabled, unwind exceptions occurring at aggregation time to the error handler when parallelProcessing is used. The default value is false for the sake of backward compatibi [...]
"id": { "kind": "attribute", "displayName": "Id", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the id of this node" },
"description": { "kind": "element", "displayName": "Description", "required": false, "type": "object", "javaType": "org.apache.camel.model.DescriptionDefinition", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the description of this node" }
}
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/MulticastDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/MulticastDefinition.java
index 11f35bc..016c805 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/MulticastDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/MulticastDefinition.java
@@ -77,7 +77,8 @@ public class MulticastDefinition extends OutputDefinition<MulticastDefinition>
@Metadata(javaType = "java.lang.Boolean")
private String parallelAggregate;
@XmlAttribute
- @Metadata(javaType = "java.lang.Boolean")
+ @Metadata(javaType = "java.lang.Boolean", deprecationNote = "Has no effect in Camel 3.x")
+ @Deprecated
private String stopOnAggregateException;
public MulticastDefinition() {
@@ -235,12 +236,13 @@ public class MulticastDefinition extends OutputDefinition<MulticastDefinition>
/**
* 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
+ * The default value is <code>false</code> for the sake of backward compatibility.
+ *
+ * @deprecated Since Camel 3.0 exceptions from the aggregate strategy are always propagated to the error handler,
+ * whether or not this option is enabled.
+ * @return the builder
*/
+ @Deprecated
public MulticastDefinition stopOnAggregateException() {
setStopOnAggregateException(Boolean.toString(true));
return this;
@@ -519,10 +521,12 @@ public class MulticastDefinition extends OutputDefinition<MulticastDefinition>
this.parallelAggregate = parallelAggregate;
}
+ @Deprecated
public String getStopOnAggregateException() {
return stopOnAggregateException;
}
+ @Deprecated
public void setStopOnAggregateException(String stopOnAggregateException) {
this.stopOnAggregateException = stopOnAggregateException;
}
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/RecipientListDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/RecipientListDefinition.java
index e381dd1..2aa911ef 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/RecipientListDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/RecipientListDefinition.java
@@ -85,7 +85,8 @@ public class RecipientListDefinition<Type extends ProcessorDefinition<Type>> ext
@Metadata(javaType = "java.lang.Boolean")
private String parallelAggregate;
@XmlAttribute
- @Metadata(javaType = "java.lang.Boolean")
+ @Metadata(javaType = "java.lang.Boolean", deprecationNote = "Has no effect in Camel 3.x")
+ @Deprecated
private String stopOnAggregateException;
public RecipientListDefinition() {
@@ -254,12 +255,14 @@ 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.
+ * The default value is <code>false</code> for the sake of backward compatibility.
+ *
+ * @deprecated Since Camel 3.0 exceptions from the aggregate strategy are always propagated to the error handler,
+ * whether or not this option is enabled.
*
- * @return the builder
+ * @return the builder
*/
+ @Deprecated
public RecipientListDefinition<Type> stopOnAggregateException() {
setStopOnAggregateException(Boolean.toString(true));
return this;
@@ -594,10 +597,12 @@ public class RecipientListDefinition<Type extends ProcessorDefinition<Type>> ext
this.parallelAggregate = parallelAggregate;
}
+ @Deprecated
public String getStopOnAggregateException() {
return stopOnAggregateException;
}
+ @Deprecated
public void setStopOnAggregateException(String stopOnAggregateException) {
this.stopOnAggregateException = stopOnAggregateException;
}
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/SplitDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/SplitDefinition.java
index 602953e..7a68864 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/SplitDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/SplitDefinition.java
@@ -77,7 +77,8 @@ public class SplitDefinition extends OutputExpressionNode implements ExecutorSer
@Metadata(javaType = "java.lang.Boolean")
private String parallelAggregate;
@XmlAttribute
- @Metadata(javaType = "java.lang.Boolean")
+ @Metadata(javaType = "java.lang.Boolean", deprecationNote = "Has no effect in Camel 3.x")
+ @Deprecated
private String stopOnAggregateException;
public SplitDefinition() {
@@ -285,24 +286,28 @@ public class SplitDefinition extends OutputExpressionNode implements ExecutorSer
/**
* 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.
+ * The default value is <code>false</code> for the sake of backward compatibility.
+ *
+ * @deprecated Since Camel 3.0 exceptions from the aggregate strategy are always propagated to the error handler,
+ * whether or not this option is enabled.
*
- * @return the builder
+ * @return the builder
*/
+ @Deprecated
public SplitDefinition stopOnAggregateException() {
return stopOnAggregateException(true);
}
/**
* 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.
+ * The default value is <code>false</code> for the sake of backward compatibility.
+ *
+ * @deprecated Since Camel 3.0 exceptions from the aggregate strategy are always propagated to the error handler,
+ * whether or not this option is enabled.
*
- * @return the builder
+ * @return the builder
*/
+ @Deprecated
public SplitDefinition stopOnAggregateException(boolean stopOnAggregateException) {
return stopOnAggregateException(Boolean.toString(stopOnAggregateException));
}
@@ -593,10 +598,12 @@ public class SplitDefinition extends OutputExpressionNode implements ExecutorSer
this.parallelAggregate = parallelAggregate;
}
+ @Deprecated
public String getStopOnAggregateException() {
return this.stopOnAggregateException;
}
+ @Deprecated
public void setStopOnAggregateException(String stopOnAggregateException) {
this.stopOnAggregateException = stopOnAggregateException;
}
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/DefaultAnnotationBasedProcessorFactory.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/DefaultAnnotationBasedProcessorFactory.java
index 29b8cb1..cde6043 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/DefaultAnnotationBasedProcessorFactory.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/DefaultAnnotationBasedProcessorFactory.java
@@ -47,7 +47,6 @@ public final class DefaultAnnotationBasedProcessorFactory implements AnnotationB
org.apache.camel.processor.RecipientList recipientList
= new org.apache.camel.processor.RecipientList(camelContext, annotation.delimiter());
recipientList.setStopOnException(annotation.stopOnException());
- recipientList.setStopOnAggregateException(annotation.stopOnAggregateException());
recipientList.setIgnoreInvalidEndpoints(annotation.ignoreInvalidEndpoints());
recipientList.setParallelProcessing(annotation.parallelProcessing());
recipientList.setParallelAggregate(annotation.parallelAggregate());
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
index 4c9dcce..6da1142 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
@@ -161,7 +161,6 @@ public class MulticastProcessor extends AsyncProcessorSupport
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;
@@ -182,21 +181,11 @@ public class MulticastProcessor extends AsyncProcessorSupport
}
public MulticastProcessor(CamelContext camelContext, Route route, 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, route, processors, aggregationStrategy, parallelProcessing, executorService, shutdownExecutorService,
- streaming, stopOnException, timeout, onPrepare,
- shareUnitOfWork, parallelAggregate, false);
- }
-
- public MulticastProcessor(CamelContext camelContext, Route route, Collection<Processor> processors,
AggregationStrategy aggregationStrategy,
boolean parallelProcessing, ExecutorService executorService, boolean shutdownExecutorService,
boolean streaming,
boolean stopOnException, long timeout, Processor onPrepare, boolean shareUnitOfWork,
- boolean parallelAggregate, boolean stopOnAggregateException) {
+ boolean parallelAggregate) {
notNull(camelContext, "camelContext");
this.camelContext = camelContext;
this.internalProcessorFactory = camelContext.adapt(ExtendedCamelContext.class).getInternalProcessorFactory();
@@ -214,7 +203,6 @@ public class MulticastProcessor extends AsyncProcessorSupport
this.onPrepare = onPrepare;
this.shareUnitOfWork = shareUnitOfWork;
this.parallelAggregate = parallelAggregate;
- this.stopOnAggregateException = stopOnAggregateException;
this.processorExchangeFactory = camelContext.adapt(ExtendedCamelContext.class)
.getProcessorExchangeFactory().newProcessorExchangeFactory(this);
}
@@ -1255,10 +1243,6 @@ public class MulticastProcessor extends AsyncProcessorSupport
return parallelAggregate;
}
- public boolean isStopOnAggregateException() {
- return stopOnAggregateException;
- }
-
public boolean isShareUnitOfWork() {
return shareUnitOfWork;
}
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientList.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientList.java
index c6c92d2..bed81851 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientList.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientList.java
@@ -59,7 +59,6 @@ public class RecipientList extends AsyncProcessorSupport implements IdAware, Rou
private final String delimiter;
private boolean parallelProcessing;
private boolean parallelAggregate;
- private boolean stopOnAggregateException;
private boolean stopOnException;
private boolean ignoreInvalidEndpoints;
private boolean streaming;
@@ -205,8 +204,7 @@ public class RecipientList extends AsyncProcessorSupport implements IdAware, Rou
recipientListProcessor = new RecipientListProcessor(
camelContext, null, expression, delimiter, producerCache, getAggregationStrategy(),
isParallelProcessing(), getExecutorService(), isShutdownExecutorService(),
- isStreaming(), isStopOnException(), getTimeout(), getOnPrepare(), isShareUnitOfWork(), isParallelAggregate(),
- isStopOnAggregateException());
+ isStreaming(), isStopOnException(), getTimeout(), getOnPrepare(), isShareUnitOfWork(), isParallelAggregate());
recipientListProcessor.setErrorHandler(errorHandler);
recipientListProcessor.setAggregateExecutorService(aggregateExecutorService);
recipientListProcessor.setIgnoreInvalidEndpoints(isIgnoreInvalidEndpoints());
@@ -274,14 +272,6 @@ public class RecipientList extends AsyncProcessorSupport implements IdAware, Rou
this.parallelAggregate = parallelAggregate;
}
- public boolean isStopOnAggregateException() {
- return stopOnAggregateException;
- }
-
- public void setStopOnAggregateException(boolean stopOnAggregateException) {
- this.stopOnAggregateException = stopOnAggregateException;
- }
-
public boolean isStopOnException() {
return stopOnException;
}
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
index 1c472d4..3a0e943 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java
@@ -160,11 +160,10 @@ public class RecipientListProcessor extends MulticastProcessor {
AggregationStrategy aggregationStrategy,
boolean parallelProcessing, ExecutorService executorService, boolean shutdownExecutorService,
boolean streaming, boolean stopOnException,
- long timeout, Processor onPrepare, boolean shareUnitOfWork, boolean parallelAggregate,
- boolean stopOnAggregateException) {
+ long timeout, Processor onPrepare, boolean shareUnitOfWork, boolean parallelAggregate) {
super(camelContext, route, null, aggregationStrategy, parallelProcessing, executorService, shutdownExecutorService,
streaming, stopOnException, timeout, onPrepare,
- shareUnitOfWork, parallelAggregate, stopOnAggregateException);
+ shareUnitOfWork, parallelAggregate);
this.expression = expression;
this.delimiter = delimiter;
this.producerCache = producerCache;
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
index b822ec8..b87adce 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
@@ -67,7 +67,7 @@ public class Splitter extends MulticastProcessor implements AsyncProcessor, Trac
boolean useSubUnitOfWork, boolean parallelAggregate) {
this(camelContext, route, expression, destination, aggregationStrategy, parallelProcessing, executorService,
shutdownExecutorService, streaming, stopOnException, timeout,
- onPrepare, useSubUnitOfWork, parallelAggregate, false, ",");
+ onPrepare, useSubUnitOfWork, parallelAggregate, ",");
}
public Splitter(CamelContext camelContext, Route route, Expression expression, Processor destination,
@@ -75,33 +75,9 @@ public class Splitter extends MulticastProcessor implements AsyncProcessor, Trac
ExecutorService executorService, boolean shutdownExecutorService, boolean streaming,
boolean stopOnException, long timeout, Processor onPrepare,
boolean useSubUnitOfWork, boolean parallelAggregate, String delimiter) {
- this(camelContext, route, expression, destination, aggregationStrategy, parallelProcessing, executorService,
- shutdownExecutorService, streaming, stopOnException, timeout,
- onPrepare, useSubUnitOfWork, parallelAggregate, false, delimiter);
- }
-
- public Splitter(CamelContext camelContext, Route route, 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, route, Collections.singleton(destination), aggregationStrategy, parallelProcessing, executorService,
- shutdownExecutorService, streaming, stopOnException,
- timeout, onPrepare, useSubUnitOfWork, parallelAggregate, stopOnAggregateException);
- this.expression = expression;
- this.delimiter = ",";
- notNull(expression, "expression");
- notNull(destination, "destination");
- }
-
- public Splitter(CamelContext camelContext, Route route, 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, String delimiter) {
super(camelContext, route, Collections.singleton(destination), aggregationStrategy, parallelProcessing, executorService,
shutdownExecutorService, streaming, stopOnException,
- timeout, onPrepare, useSubUnitOfWork, parallelAggregate, stopOnAggregateException);
+ timeout, onPrepare, useSubUnitOfWork, parallelAggregate);
this.expression = expression;
StringHelper.notEmpty(delimiter, "delimiter");
this.delimiter = delimiter;
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/MulticastReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/MulticastReifier.java
index aa0962b..4209e5b 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/MulticastReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/MulticastReifier.java
@@ -62,7 +62,6 @@ public class MulticastReifier extends ProcessorReifier<MulticastDefinition> {
boolean isStreaming = parseBoolean(definition.getStreaming(), false);
boolean isStopOnException = parseBoolean(definition.getStopOnException(), false);
boolean isParallelAggregate = parseBoolean(definition.getParallelAggregate(), false);
- boolean isStopOnAggregateException = parseBoolean(definition.getStopOnAggregateException(), false);
boolean shutdownThreadPool = willCreateNewThreadPool(definition, isParallelProcessing);
ExecutorService threadPool = getConfiguredExecutorService("Multicast", definition, isParallelProcessing);
@@ -77,8 +76,7 @@ public class MulticastReifier extends ProcessorReifier<MulticastDefinition> {
MulticastProcessor answer = new MulticastProcessor(
camelContext, route, list, strategy, isParallelProcessing, threadPool, shutdownThreadPool, isStreaming,
- isStopOnException, timeout, definition.getOnPrepare(), isShareUnitOfWork, isParallelAggregate,
- isStopOnAggregateException);
+ isStopOnException, timeout, definition.getOnPrepare(), isShareUnitOfWork, isParallelAggregate);
return answer;
}
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RecipientListReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RecipientListReifier.java
index 2c1793f..41f223a 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RecipientListReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RecipientListReifier.java
@@ -51,7 +51,6 @@ public class RecipientListReifier extends ProcessorReifier<RecipientListDefiniti
boolean isShareUnitOfWork = parseBoolean(definition.getShareUnitOfWork(), false);
boolean isStopOnException = parseBoolean(definition.getStopOnException(), false);
boolean isIgnoreInvalidEndpoints = parseBoolean(definition.getIgnoreInvalidEndpoints(), false);
- boolean isStopOnAggregateException = parseBoolean(definition.getStopOnAggregateException(), false);
RecipientList answer;
String delimiter = parseString(definition.getDelimiter());
@@ -67,7 +66,6 @@ public class RecipientListReifier extends ProcessorReifier<RecipientListDefiniti
answer.setShareUnitOfWork(isShareUnitOfWork);
answer.setStopOnException(isStopOnException);
answer.setIgnoreInvalidEndpoints(isIgnoreInvalidEndpoints);
- answer.setStopOnAggregateException(isStopOnAggregateException);
Integer num = parseInt(definition.getCacheSize());
if (num != null) {
answer.setCacheSize(num);
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java
index 90ede30..a6fbf55 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java
@@ -46,7 +46,6 @@ public class SplitReifier extends ExpressionReifier<SplitDefinition> {
boolean isStreaming = parseBoolean(definition.getStreaming(), false);
boolean isShareUnitOfWork = parseBoolean(definition.getShareUnitOfWork(), false);
boolean isParallelAggregate = parseBoolean(definition.getParallelAggregate(), false);
- boolean isStopOnAggregateException = parseBoolean(definition.getStopOnAggregateException(), false);
boolean isStopOnException = parseBoolean(definition.getStopOnException(), false);
boolean shutdownThreadPool = willCreateNewThreadPool(definition, isParallelProcessing);
ExecutorService threadPool = getConfiguredExecutorService("Split", definition, isParallelProcessing);
@@ -68,12 +67,12 @@ public class SplitReifier extends ExpressionReifier<SplitDefinition> {
answer = new Splitter(
camelContext, route, exp, childProcessor, definition.getAggregationStrategy(), isParallelProcessing,
threadPool, shutdownThreadPool, isStreaming, isStopOnException, timeout, definition.getOnPrepare(),
- isShareUnitOfWork, isParallelAggregate, isStopOnAggregateException, delimiter);
+ isShareUnitOfWork, isParallelAggregate, delimiter);
} else {
answer = new Splitter(
camelContext, route, exp, childProcessor, definition.getAggregationStrategy(), isParallelProcessing,
threadPool, shutdownThreadPool, isStreaming, isStopOnException, timeout, definition.getOnPrepare(),
- isShareUnitOfWork, isParallelAggregate, isStopOnAggregateException);
+ isShareUnitOfWork, isParallelAggregate);
}
return answer;
diff --git a/core/camel-core/src/test/java/org/apache/camel/issues/MulticastParallelWithAggregationStrategyThrowingExceptionTest.java b/core/camel-core/src/test/java/org/apache/camel/issues/MulticastParallelWithAggregationStrategyThrowingExceptionTest.java
index e16fdd1..1a96c37 100644
--- a/core/camel-core/src/test/java/org/apache/camel/issues/MulticastParallelWithAggregationStrategyThrowingExceptionTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/issues/MulticastParallelWithAggregationStrategyThrowingExceptionTest.java
@@ -49,7 +49,7 @@ public class MulticastParallelWithAggregationStrategyThrowingExceptionTest exten
// 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()
+ from("direct:start").multicast(new MyAggregateBean()).parallelProcessing()
.shareUnitOfWork().to("mock:a").to("mock:b").end()
.to("mock:end");
}
@@ -60,7 +60,11 @@ public class MulticastParallelWithAggregationStrategyThrowingExceptionTest exten
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
- throw new RuntimeException("Simulating a runtime exception thrown from the aggregation strategy");
+ if (oldExchange != null) {
+ throw new RuntimeException("Simulating a runtime exception thrown from the aggregation strategy");
+ } else {
+ return newExchange;
+ }
}
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/issues/RecipientListParallelWithAggregationStrategyThrowingExceptionTest.java b/core/camel-core/src/test/java/org/apache/camel/issues/RecipientListParallelWithAggregationStrategyThrowingExceptionTest.java
index 459febe..f09d4f0 100644
--- a/core/camel-core/src/test/java/org/apache/camel/issues/RecipientListParallelWithAggregationStrategyThrowingExceptionTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/issues/RecipientListParallelWithAggregationStrategyThrowingExceptionTest.java
@@ -50,7 +50,7 @@ public class RecipientListParallelWithAggregationStrategyThrowingExceptionTest e
// exceptions
// from the aggregation strategy also.
from("direct:start").recipientList(header("recipients")).aggregationStrategy(new MyAggregateBean())
- .parallelProcessing().stopOnAggregateException()
+ .parallelProcessing()
.shareUnitOfWork().end().to("mock:end");
}
};
@@ -60,7 +60,11 @@ public class RecipientListParallelWithAggregationStrategyThrowingExceptionTest e
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
- throw new RuntimeException("Simulating a runtime exception thrown from the aggregation strategy");
+ if (oldExchange != null) {
+ throw new RuntimeException("Simulating a runtime exception thrown from the aggregation strategy");
+ } else {
+ return newExchange;
+ }
}
}
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelWithAggregationStrategyThrowingExceptionTest.java b/core/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelWithAggregationStrategyThrowingExceptionTest.java
index 44273a2..ceba5f9 100644
--- a/core/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelWithAggregationStrategyThrowingExceptionTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelWithAggregationStrategyThrowingExceptionTest.java
@@ -49,7 +49,7 @@ public class SplitterParallelWithAggregationStrategyThrowingExceptionTest extend
// exceptions
// from the aggregation strategy also.
from("direct:start").split(body().tokenize("@")).aggregationStrategy(new MyAggregateBean()).parallelProcessing()
- .stopOnAggregateException().shareUnitOfWork()
+ .shareUnitOfWork()
.to("mock:a").end().to("mock:end");
}
};
@@ -59,7 +59,11 @@ public class SplitterParallelWithAggregationStrategyThrowingExceptionTest extend
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
- throw new RuntimeException("Simulating a runtime exception thrown from the aggregation strategy");
+ if (oldExchange != null) {
+ throw new RuntimeException("Simulating a runtime exception thrown from the aggregation strategy");
+ } else {
+ return newExchange;
+ }
}
}
}