camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject git commit: CAMEL-6246: Added optimistick lock retry policy to aggregate eip. Thanks to Aaron Whiteside for the patch.
Date Sat, 11 May 2013 09:07:36 GMT
Updated Branches:
  refs/heads/master 96ce8d0f8 -> fabc86825


CAMEL-6246: Added optimistick lock retry policy to aggregate eip. Thanks to Aaron Whiteside
for the patch.


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

Branch: refs/heads/master
Commit: fabc8682592f1169b006e1de85840eb758b8fadb
Parents: 96ce8d0
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Sat May 11 11:05:12 2013 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Sat May 11 11:05:12 2013 +0200

----------------------------------------------------------------------
 .../apache/camel/model/AggregateDefinition.java    |   34 ++++-
 .../model/OptimisticLockRetryPolicyDefinition.java |  140 ++++++++++++++
 .../processor/aggregate/AggregateProcessor.java    |   27 ++-
 .../aggregate/OptimisticLockRetryPolicy.java       |  148 ++++++++++++++
 .../resources/org/apache/camel/model/jaxb.index    |    1 +
 ...DistributedConcurrentPerCorrelationKeyTest.java |    2 -
 .../DistributedOptimisticLockFailingTest.java      |  149 +++++++++++++++
 .../aggregator/OptimisticLockRetryPolicyTest.java  |  122 ++++++++++++
 8 files changed, 612 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/fabc8682/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java b/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
index d8f26d8..fce58e5 100644
--- a/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
@@ -38,6 +38,7 @@ import org.apache.camel.processor.UnitOfWorkProcessor;
 import org.apache.camel.processor.aggregate.AggregateProcessor;
 import org.apache.camel.processor.aggregate.AggregationStrategy;
 import org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy;
+import org.apache.camel.processor.aggregate.OptimisticLockRetryPolicy;
 import org.apache.camel.spi.AggregationRepository;
 import org.apache.camel.spi.RouteContext;
 import org.apache.camel.util.concurrent.SynchronousExecutorService;
@@ -58,6 +59,8 @@ public class AggregateDefinition extends ProcessorDefinition<AggregateDefinition
     private ExpressionSubElementDefinition completionTimeoutExpression;
     @XmlElement(name = "completionSize")
     private ExpressionSubElementDefinition completionSizeExpression;
+    @XmlElement(name = "optimisticLockRetryPolicy")
+    private OptimisticLockRetryPolicyDefinition optimisticLockRetryPolicyDefinition;
     @XmlTransient
     private ExpressionDefinition expression;
     @XmlElementRef
@@ -70,6 +73,8 @@ public class AggregateDefinition extends ProcessorDefinition<AggregateDefinition
     private ScheduledExecutorService timeoutCheckerExecutorService;
     @XmlTransient
     private AggregationRepository aggregationRepository;
+    @XmlTransient
+    private OptimisticLockRetryPolicy optimisticLockRetryPolicy;
     @XmlAttribute
     private Boolean parallelProcessing;
     @XmlAttribute
@@ -237,7 +242,13 @@ public class AggregateDefinition extends ProcessorDefinition<AggregateDefinition
         if (getForceCompletionOnStop() != null) {
             answer.setForceCompletionOnStop(getForceCompletionOnStop());
         }
-
+        if (optimisticLockRetryPolicy == null) {
+            if (getOptimisticLockRetryPolicyDefinition() != null) {
+                answer.setOptimisticLockRetryPolicy(getOptimisticLockRetryPolicyDefinition().createOptimisticLockRetryPolicy());
+            }
+        } else {
+            answer.setOptimisticLockRetryPolicy(optimisticLockRetryPolicy);
+        }
         return answer;
     }
 
@@ -316,6 +327,22 @@ public class AggregateDefinition extends ProcessorDefinition<AggregateDefinition
         this.completionSize = completionSize;
     }
 
+    public OptimisticLockRetryPolicyDefinition getOptimisticLockRetryPolicyDefinition() {
+        return optimisticLockRetryPolicyDefinition;
+    }
+
+    public void setOptimisticLockRetryPolicyDefinition(OptimisticLockRetryPolicyDefinition
optimisticLockRetryPolicyDefinition) {
+        this.optimisticLockRetryPolicyDefinition = optimisticLockRetryPolicyDefinition;
+    }
+
+    public OptimisticLockRetryPolicy getOptimisticLockRetryPolicy() {
+        return optimisticLockRetryPolicy;
+    }
+
+    public void setOptimisticLockRetryPolicy(OptimisticLockRetryPolicy optimisticLockRetryPolicy)
{
+        this.optimisticLockRetryPolicy = optimisticLockRetryPolicy;
+    }
+
     public Long getCompletionInterval() {
         return completionInterval;
     }
@@ -738,6 +765,11 @@ public class AggregateDefinition extends ProcessorDefinition<AggregateDefinition
         setOptimisticLocking(true);
         return this;
     }
+
+    public AggregateDefinition optimisticLockRetryPolicy(OptimisticLockRetryPolicy policy)
{
+        setOptimisticLockRetryPolicy(policy);
+        return this;
+    }
     
     public AggregateDefinition executorService(ExecutorService executorService) {
         setExecutorService(executorService);

http://git-wip-us.apache.org/repos/asf/camel/blob/fabc8682/camel-core/src/main/java/org/apache/camel/model/OptimisticLockRetryPolicyDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/OptimisticLockRetryPolicyDefinition.java
b/camel-core/src/main/java/org/apache/camel/model/OptimisticLockRetryPolicyDefinition.java
new file mode 100644
index 0000000..87b449d
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/model/OptimisticLockRetryPolicyDefinition.java
@@ -0,0 +1,140 @@
+/**
+ * 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.model;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import org.apache.camel.processor.aggregate.OptimisticLockRetryPolicy;
+
+/**
+ * Represents an XML &lt;optimisticLockRetryPolicy/&gt; element
+ *
+ * @version
+ */
+@XmlRootElement(name = "optimisticLockRetryPolicy")
+@XmlAccessorType(XmlAccessType.FIELD)
+public class OptimisticLockRetryPolicyDefinition {
+    @XmlAttribute
+    private Integer maximumRetries;
+    @XmlAttribute
+    private Long retryDelay;
+    @XmlAttribute
+    private Long maximumRetryDelay;
+    @XmlAttribute
+    private Boolean exponentialBackOff;
+    @XmlAttribute
+    private Boolean randomBackOff;
+
+    public OptimisticLockRetryPolicyDefinition() {
+    }
+
+    public OptimisticLockRetryPolicy createOptimisticLockRetryPolicy() {
+        OptimisticLockRetryPolicy policy = new OptimisticLockRetryPolicy();
+        if (maximumRetries != null) {
+            policy.setMaximumRetries(maximumRetries);
+        }
+        if (retryDelay != null) {
+            policy.setRetryDelay(retryDelay);
+        }
+        if (maximumRetryDelay != null) {
+            policy.setMaximumRetryDelay(maximumRetryDelay);
+        }
+        if (exponentialBackOff != null) {
+            policy.setExponentialBackOff(exponentialBackOff);
+        }
+        if (randomBackOff != null) {
+            policy.setRandomBackOff(randomBackOff);
+        }
+        return policy;
+    }
+
+    public OptimisticLockRetryPolicyDefinition maximumRetries(int maximumRetries) {
+        setMaximumRetries(maximumRetries);
+        return this;
+    }
+
+    public Integer getMaximumRetries() {
+        return maximumRetries;
+    }
+
+    public void setMaximumRetries(Integer maximumRetries) {
+        this.maximumRetries = maximumRetries;
+    }
+
+    public OptimisticLockRetryPolicyDefinition retryDelay(long retryDelay) {
+        setRetryDelay(retryDelay);
+        return this;
+    }
+
+    public Long getRetryDelay() {
+        return retryDelay;
+    }
+
+    public void setRetryDelay(Long retryDelay) {
+        this.retryDelay = retryDelay;
+    }
+
+    public OptimisticLockRetryPolicyDefinition maximumRetryDelay(long maximumRetryDelay)
{
+        setMaximumRetryDelay(maximumRetryDelay);
+        return this;
+    }
+
+    public Long getMaximumRetryDelay() {
+        return maximumRetryDelay;
+    }
+
+    public void setMaximumRetryDelay(Long maximumRetryDelay) {
+        this.maximumRetryDelay = maximumRetryDelay;
+    }
+
+    public OptimisticLockRetryPolicyDefinition exponentialBackOff() {
+        return exponentialBackOff(true);
+    }
+
+    public OptimisticLockRetryPolicyDefinition exponentialBackOff(boolean exponentialBackOff)
{
+        setExponentialBackOff(exponentialBackOff);
+        return this;
+    }
+
+    public Boolean getExponentialBackOff() {
+        return exponentialBackOff;
+    }
+
+    public void setExponentialBackOff(Boolean exponentialBackOff) {
+        this.exponentialBackOff = exponentialBackOff;
+    }
+
+    public OptimisticLockRetryPolicyDefinition randomBackOff() {
+        return randomBackOff(true);
+    }
+
+    public OptimisticLockRetryPolicyDefinition randomBackOff(boolean randomBackOff) {
+        setRandomBackOff(randomBackOff);
+        return this;
+    }
+
+    public Boolean getRandomBackOff() {
+        return randomBackOff;
+    }
+
+    public void setRandomBackOff(Boolean randomBackOff) {
+        this.randomBackOff = randomBackOff;
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/fabc8682/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
b/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
index c113c60..46ef290 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
@@ -21,7 +21,6 @@ import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentSkipListSet;
@@ -90,6 +89,7 @@ public class AggregateProcessor extends ServiceSupport implements Processor,
Nav
     private final Expression correlationExpression;
     private final ExecutorService executorService;
     private final boolean shutdownExecutorService;
+    private OptimisticLockRetryPolicy optimisticLockRetryPolicy = new OptimisticLockRetryPolicy();
     private ScheduledExecutorService timeoutCheckerExecutorService;
     private boolean shutdownTimeoutCheckerExecutorService;
     private ScheduledExecutorService recoverService;
@@ -194,24 +194,27 @@ public class AggregateProcessor extends ServiceSupport implements Processor,
Nav
 
         // when optimist locking is enabled we keep trying until we succeed
         if (optimisticLocking) {
-            boolean done = false;
+            boolean exhaustedRetries = true;
             int attempt = 0;
-            while (!done) {
+            do {
                 attempt++;
                 // copy exchange, and do not share the unit of work
                 // the aggregated output runs in another unit of work
                 Exchange copy = ExchangeHelper.createCorrelatedCopy(exchange, false);
                 try {
                     doAggregation(key, copy);
-                    done = true;
+                    exhaustedRetries = false;
+                    break;
                 } catch (OptimisticLockingAggregationRepository.OptimisticLockingException
e) {
                     LOG.trace("On attempt {} OptimisticLockingAggregationRepository: {} threw
OptimisticLockingException while trying to add() key: {} and exchange: {}",
                               new Object[]{attempt, aggregationRepository, key, copy, e});
+                    optimisticLockRetryPolicy.doDelay(attempt);
                 }
-                // use a little random delay to avoid being to aggressive when retrying,
and avoid potential clashing
-                int ran = new Random().nextInt(1000);
-                LOG.trace("Sleeping for {} millis before attempting again", ran);
-                Thread.sleep(ran);
+            } while (optimisticLockRetryPolicy.shouldRetry(attempt));
+
+            if (exhaustedRetries) {
+                throw new CamelExchangeException("Exhausted optimistic locking retry attempts,
tried " + attempt + " times", exchange,
+                        new OptimisticLockingAggregationRepository.OptimisticLockingException());
             }
         } else {
             // copy exchange, and do not share the unit of work
@@ -689,6 +692,14 @@ public class AggregateProcessor extends ServiceSupport implements Processor,
Nav
         this.shutdownTimeoutCheckerExecutorService = shutdownTimeoutCheckerExecutorService;
     }
 
+    public void setOptimisticLockRetryPolicy(OptimisticLockRetryPolicy optimisticLockRetryPolicy)
{
+        this.optimisticLockRetryPolicy = optimisticLockRetryPolicy;
+    }
+
+    public OptimisticLockRetryPolicy getOptimisticLockRetryPolicy() {
+        return optimisticLockRetryPolicy;
+    }
+
     /**
      * On completion task which keeps the booking of the in progress up to date
      */

http://git-wip-us.apache.org/repos/asf/camel/blob/fabc8682/camel-core/src/main/java/org/apache/camel/processor/aggregate/OptimisticLockRetryPolicy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/aggregate/OptimisticLockRetryPolicy.java
b/camel-core/src/main/java/org/apache/camel/processor/aggregate/OptimisticLockRetryPolicy.java
new file mode 100644
index 0000000..3ca3081
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/aggregate/OptimisticLockRetryPolicy.java
@@ -0,0 +1,148 @@
+/**
+ * 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.processor.aggregate;
+
+import java.util.Random;
+
+/**
+ * Class to control how failed optimistic locks are tried. This policy supports random and
exponential back-off delays.
+ * <p/>
+ * If {@code randomBackOff} is enabled and a value is supplied for {@code retryDelay} the
value will be ignored.
+ * <p/>
+ * If {@code randomBackOff} is enabled and no value is set for {@code maximumRetryDelay},
a default value of 1000ms will
+ * be used, the random delay will be between 0 and 1000 milliseconds.
+ * <p/>
+ * If both {@code randomBackOff} and {@code exponentialBackOff} are enabled, {@code exponentialBackOff}
will take precedence.
+ * <p/>
+ * If {@code exponentialBackOff} is enabled and a value is set for {@code maximumRetryDelay},
the retry delay will keep
+ * doubling in value until it reaches or exceeds {@code maximumRetryDelay}. After it has
reached or exceeded {@code maximumRetryDelay}
+ * the value of {@code maximumRetryDelay} will be used as the retry delay.
+ * <p/>
+ * If both {@code exponentialBackOff} and {@code randomBackOff} are disabled, the value of
{@code retryDelay} will be used
+ * as the retry delay and remain constant through all the retry attempts.
+ * <p/>
+ * If the value of {@code maximumRetries} is set above zero, retry attempts will stop at
the value specified.
+ * <p/>
+ * The default behaviour of this policy is to retry forever and exponentially increase the
back-off delay starting with 50ms.
+ *
+ * @version
+ */
+public class OptimisticLockRetryPolicy {
+    private static final long DEFAULT_MAXIMUM_RETRY_DELAY = 1000L;
+
+    private int maximumRetries;
+    private long retryDelay = 50L;
+    private long maximumRetryDelay;
+    private boolean exponentialBackOff = true;
+    private boolean randomBackOff;
+
+    public OptimisticLockRetryPolicy() {
+    }
+
+    public boolean shouldRetry(final int retryCounter) {
+        return maximumRetries <= 0 || retryCounter < maximumRetries;
+    }
+
+    public void doDelay(final int retryCounter) throws InterruptedException {
+        if (retryDelay > 0 || randomBackOff) {
+            long sleepFor;
+            sleepFor = exponentialBackOff ? (retryDelay << retryCounter)
+                    : (randomBackOff ? new Random().nextInt((int)(maximumRetryDelay >
0 ? maximumRetryDelay : DEFAULT_MAXIMUM_RETRY_DELAY)) : retryDelay);
+            if (maximumRetryDelay > 0 && sleepFor > maximumRetryDelay) {
+                sleepFor = maximumRetryDelay;
+            }
+            Thread.sleep(sleepFor);
+        }
+    }
+
+    public int getMaximumRetries() {
+        return maximumRetries;
+    }
+
+    public void setMaximumRetries(int maximumRetries) {
+        this.maximumRetries = maximumRetries;
+    }
+
+    public OptimisticLockRetryPolicy maximumRetries(int maximumRetries) {
+        setMaximumRetries(maximumRetries);
+        return this;
+    }
+
+    public long getRetryDelay() {
+        return retryDelay;
+    }
+
+    public void setRetryDelay(long retryDelay) {
+        this.retryDelay = retryDelay;
+    }
+
+    public OptimisticLockRetryPolicy retryDelay(long retryDelay) {
+        setRetryDelay(retryDelay);
+        return this;
+    }
+
+    public long getMaximumRetryDelay() {
+        return maximumRetryDelay;
+    }
+
+    public void setMaximumRetryDelay(long maximumRetryDelay) {
+        this.maximumRetryDelay = maximumRetryDelay;
+    }
+
+    public OptimisticLockRetryPolicy maximumRetryDelay(long maximumRetryDelay) {
+        setMaximumRetryDelay(maximumRetryDelay);
+        return this;
+    }
+
+    public boolean isExponentialBackOff() {
+        return exponentialBackOff;
+    }
+
+    public void setExponentialBackOff(boolean exponentialBackOff) {
+        this.exponentialBackOff = exponentialBackOff;
+    }
+
+    public OptimisticLockRetryPolicy exponentialBackOff() {
+        setExponentialBackOff(true);
+        return this;
+    }
+
+    public boolean isRandomBackOff() {
+        return randomBackOff;
+    }
+
+    public void setRandomBackOff(boolean randomBackOff) {
+        this.randomBackOff = randomBackOff;
+    }
+
+    public OptimisticLockRetryPolicy randomBackOff() {
+        setRandomBackOff(true);
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("OptimisticLockRetryPolicy[");
+        sb.append("maximumRetries=").append(maximumRetries);
+        sb.append(", retryDelay=").append(retryDelay);
+        sb.append(", maximumRetryDelay=").append(maximumRetryDelay);
+        sb.append(", exponentialBackOff=").append(exponentialBackOff);
+        sb.append(", randomBackOff=").append(randomBackOff);
+        sb.append(']');
+        return sb.toString();
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/fabc8682/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
----------------------------------------------------------------------
diff --git a/camel-core/src/main/resources/org/apache/camel/model/jaxb.index b/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
index a877af7..1e32f42 100644
--- a/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
+++ b/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
@@ -44,6 +44,7 @@ MarshalDefinition
 MulticastDefinition
 OnCompletionDefinition
 OnExceptionDefinition
+OptimisticLockRetryPolicyDefinition
 OptionalIdentifiedDefinition
 OtherwiseDefinition
 PackageScanDefinition

http://git-wip-us.apache.org/repos/asf/camel/blob/fabc8682/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedConcurrentPerCorrelationKeyTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedConcurrentPerCorrelationKeyTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedConcurrentPerCorrelationKeyTest.java
index 4d089a7..44caad2 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedConcurrentPerCorrelationKeyTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedConcurrentPerCorrelationKeyTest.java
@@ -27,7 +27,6 @@ import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.component.mock.MockEndpoint;
 import org.apache.camel.processor.BodyInAggregatingStrategy;
 import org.apache.camel.processor.aggregate.MemoryAggregationRepository;
-import org.junit.Test;
 
 /**
  * @version
@@ -39,7 +38,6 @@ public class DistributedConcurrentPerCorrelationKeyTest extends AbstractDistribu
     private int size = 200;
     private final String uri = "direct:start";
 
-    @Test
     public void testAggregateConcurrentPerCorrelationKey() throws Exception {
         ExecutorService service = Executors.newFixedThreadPool(50);
         List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();

http://git-wip-us.apache.org/repos/asf/camel/blob/fabc8682/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedOptimisticLockFailingTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedOptimisticLockFailingTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedOptimisticLockFailingTest.java
new file mode 100644
index 0000000..810eb18
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/DistributedOptimisticLockFailingTest.java
@@ -0,0 +1,149 @@
+/**
+ * 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.processor.aggregator;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.BodyInAggregatingStrategy;
+import org.apache.camel.processor.aggregate.MemoryAggregationRepository;
+import org.apache.camel.processor.aggregate.OptimisticLockRetryPolicy;
+import org.apache.camel.spi.OptimisticLockingAggregationRepository;
+
+/**
+ * @version
+ */
+public class DistributedOptimisticLockFailingTest extends AbstractDistributedTest {
+
+    private static final class AlwaysFailingRepository extends MemoryAggregationRepository
{
+        @Override
+        public Exchange add(CamelContext camelContext, String key, Exchange oldExchange,
Exchange newExchange) {
+            throw new OptimisticLockingException();
+        }
+    }
+
+    private static final class EverySecondOneFailsRepository extends MemoryAggregationRepository
{
+        private AtomicInteger counter = new AtomicInteger();
+        private EverySecondOneFailsRepository() {
+            super(true);
+        }
+        @Override
+        public Exchange add(CamelContext camelContext, String key, Exchange oldExchange,
Exchange newExchange) {
+            int count = counter.incrementAndGet();
+            if (count % 2 == 0) {
+                throw new OptimisticLockingException();
+            } else {
+                return super.add(camelContext, key, oldExchange, newExchange);
+            }
+        }
+    }
+    private EverySecondOneFailsRepository sharedRepository = new EverySecondOneFailsRepository();
+
+    public void testAlwaysFails() throws Exception {
+        MockEndpoint mock = getMockEndpoint("mock:result");
+        mock.expectedMessageCount(0);
+        MockEndpoint mock2 = getMockEndpoint2("mock:result");
+        mock2.expectedMessageCount(0);
+
+        try {
+            template.sendBodyAndHeader("direct:fails", "hello world", "id", 1);
+            fail("Should throw CamelExecutionException");
+        } catch (CamelExecutionException e) {
+            assertIsInstanceOf(CamelExchangeException.class, e.getCause());
+            assertIsInstanceOf(OptimisticLockingAggregationRepository.OptimisticLockingException.class,
e.getCause().getCause());
+        }
+
+        try {
+            template2.sendBodyAndHeader("direct:fails", "hello world", "id", 1);
+            fail("Should throw CamelExecutionException");
+        } catch (CamelExecutionException e) {
+            assertIsInstanceOf(CamelExchangeException.class, e.getCause());
+            assertIsInstanceOf(OptimisticLockingAggregationRepository.OptimisticLockingException.class,
e.getCause().getCause());
+        }
+
+        mock.assertIsSatisfied();
+        mock2.assertIsSatisfied();
+    }
+
+    public void testEverySecondOneFails() throws Exception {
+        int size = 200;
+        ExecutorService service = Executors.newFixedThreadPool(50);
+        List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();
+        for (int i = 0; i < size; i++) {
+            final int id = i % 25;
+            final int choice = i % 2;
+            final int count = i;
+            tasks.add(new Callable<Object>() {
+                public Object call() throws Exception {
+                    if (choice == 0) {
+                        template.sendBodyAndHeader("direct:everysecondone", "" + count, "id",
id);
+                    } else {
+                        template2.sendBodyAndHeader("direct:everysecondone", "" + count,
"id", id);
+                    }
+                    return null;
+                }
+            });
+        }
+
+        MockEndpoint mock = getMockEndpoint("mock:result");
+        MockEndpoint mock2 = getMockEndpoint2("mock:result");
+
+        // submit all tasks
+        service.invokeAll(tasks);
+        service.shutdown();
+        service.awaitTermination(10, TimeUnit.SECONDS);
+
+        int contextCount = mock.getReceivedCounter();
+        int context2Count = mock2.getReceivedCounter();
+
+        assertEquals(25, contextCount + context2Count);
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("direct:fails")
+                    .aggregate(header("id"), new BodyInAggregatingStrategy())
+                        .aggregationRepository(new AlwaysFailingRepository())
+                        .optimisticLocking()
+                        .optimisticLockRetryPolicy(new OptimisticLockRetryPolicy().maximumRetries(5))
+                        .completionSize(2)
+                        .to("mock:result");
+
+                from("direct:everysecondone")
+                    .aggregate(header("id"), new BodyInAggregatingStrategy())
+                        .aggregationRepository(sharedRepository)
+                        .optimisticLocking()
+                        .completionSize(8)
+                        .to("mock:result");
+            }
+        };
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/fabc8682/camel-core/src/test/java/org/apache/camel/processor/aggregator/OptimisticLockRetryPolicyTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/OptimisticLockRetryPolicyTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/OptimisticLockRetryPolicyTest.java
new file mode 100644
index 0000000..cbea08d
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/OptimisticLockRetryPolicyTest.java
@@ -0,0 +1,122 @@
+/**
+ * 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.processor.aggregator;
+
+import junit.framework.TestCase;
+import org.apache.camel.processor.aggregate.OptimisticLockRetryPolicy;
+
+/**
+ * @version
+ */
+public class OptimisticLockRetryPolicyTest extends TestCase {
+
+    private static long precision = 5L; // give or take 5ms
+
+    public void testRandomBackOff() throws Exception {
+        OptimisticLockRetryPolicy policy = new OptimisticLockRetryPolicy();
+        policy.setRandomBackOff(true);
+        policy.setExponentialBackOff(false);
+        policy.setMaximumRetryDelay(500L);
+
+        for (int i = 0; i < 10; i++) {
+            long start = System.currentTimeMillis();
+            policy.doDelay(i);
+            long elapsed = System.currentTimeMillis() - start;
+            assertTrue(elapsed <= policy.getMaximumRetryDelay() + precision &&
elapsed >= 0);
+        }
+    }
+
+    public void testExponentialBackOff() throws Exception {
+        OptimisticLockRetryPolicy policy = new OptimisticLockRetryPolicy();
+        policy.setRandomBackOff(false);
+        policy.setExponentialBackOff(true);
+        policy.setMaximumRetryDelay(0L);
+        policy.setRetryDelay(50L);
+
+        for (int i = 0; i < 10; i++) {
+            long start = System.currentTimeMillis();
+            policy.doDelay(i);
+            long elapsed = System.currentTimeMillis() - start;
+            assertTrue(elapsed >= (policy.getRetryDelay() << i) - precision);
+            assertTrue(elapsed <= (policy.getRetryDelay() << i) + precision);
+        }
+    }
+
+    public void testExponentialBackOffMaximumRetryDelay() throws Exception {
+        OptimisticLockRetryPolicy policy = new OptimisticLockRetryPolicy();
+        policy.setRandomBackOff(false);
+        policy.setExponentialBackOff(true);
+        policy.setMaximumRetryDelay(200L);
+        policy.setRetryDelay(50L);
+
+        for (int i = 0; i < 10; i++) {
+            long start = System.currentTimeMillis();
+            policy.doDelay(i);
+            long elapsed = System.currentTimeMillis() - start;
+            switch (i) {
+            case 0:
+                assertTrue(elapsed <= 50 + precision);
+                assertTrue(elapsed >= 50 - precision);
+                break;
+            case 1:
+                assertTrue(elapsed <= 100 + precision);
+                assertTrue(elapsed >= 100 - precision);
+                break;
+            default:
+                assertTrue(elapsed <= 200 + precision);
+                assertTrue(elapsed >= 200 - precision);
+                break;
+            }
+        }
+    }
+
+    public void testRetryDelay() throws Exception {
+        OptimisticLockRetryPolicy policy = new OptimisticLockRetryPolicy();
+        policy.setRandomBackOff(false);
+        policy.setExponentialBackOff(false);
+        policy.setMaximumRetryDelay(0L);
+        policy.setRetryDelay(50L);
+
+        for (int i = 0; i < 10; i++) {
+            long start = System.currentTimeMillis();
+            policy.doDelay(i);
+            long elapsed = System.currentTimeMillis() - start;
+            assertTrue(elapsed <= policy.getRetryDelay() + precision);
+            assertTrue(elapsed >= policy.getRetryDelay() - precision);
+        }
+    }
+
+    public void testMaximumRetries() throws Exception {
+        OptimisticLockRetryPolicy policy = new OptimisticLockRetryPolicy();
+        policy.setRandomBackOff(false);
+        policy.setExponentialBackOff(false);
+        policy.setMaximumRetryDelay(0L);
+        policy.setMaximumRetries(2);
+        policy.setRetryDelay(50L);
+
+        for (int i = 0; i < 10; i++) {
+            switch (i) {
+            case 0:
+            case 1:
+                assertTrue(policy.shouldRetry(i));
+                break;
+            default:
+                assertFalse(policy.shouldRetry(i));
+            }
+        }
+    }
+}
\ No newline at end of file


Mime
View raw message