Author: davsclaus
Date: Fri Mar 1 09:48:39 2013
New Revision: 1451546
URL: http://svn.apache.org/r1451546
Log:
CAMEL-6097: Fixed race condition in aggregate eip when doing recovery task.
Modified:
camel/branches/camel-2.10.x/ (props changed)
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Merged /camel/trunk:r1451544
Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified:
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
URL:
http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java?rev=1451546&r1=1451545&r2=1451546&view=diff
==============================================================================
---
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
(original)
+++
camel/branches/camel-2.10.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
Fri Mar 1 09:48:39 2013
@@ -20,6 +20,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -756,6 +757,9 @@ public class AggregateProcessor extends
LOG.trace("Starting recover check");
+ // copy the current in progress before doing scan
+ final Set<String> copyOfInProgress = new
LinkedHashSet<String>(inProgressCompleteExchanges);
+
Set<String> exchangeIds = recoverable.scan(camelContext);
for (String exchangeId : exchangeIds) {
@@ -765,7 +769,9 @@ public class AggregateProcessor extends
return;
}
- boolean inProgress =
inProgressCompleteExchanges.contains(exchangeId);
+ // consider in progress if it was in progress before we did
the scan, or currently after we did the scan
+ // its safer to consider it in progress than risk duplicates
due both in progress + recovered
+ boolean inProgress = copyOfInProgress.contains(exchangeId) ||
inProgressCompleteExchanges.contains(exchangeId);
if (inProgress) {
LOG.trace("Aggregated exchange with id: {} is already in
progress.", exchangeId);
} else {
