camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ningji...@apache.org
Subject git commit: CAMEL-7973 Fixed the issue that CircuitBreakerLoadBalancer fails on async processors with thanks to Matteo
Date Thu, 30 Oct 2014 09:04:57 GMT
Repository: camel
Updated Branches:
  refs/heads/master 76c3a420c -> 799b45df0


CAMEL-7973 Fixed the issue that CircuitBreakerLoadBalancer fails on async processors with
thanks to Matteo


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

Branch: refs/heads/master
Commit: 799b45df0c2e670f0240ddd3e9c6256fcb4b3740
Parents: 76c3a42
Author: Willem Jiang <willem.jiang@gmail.com>
Authored: Thu Oct 30 17:04:28 2014 +0800
Committer: Willem Jiang <willem.jiang@gmail.com>
Committed: Thu Oct 30 17:04:28 2014 +0800

----------------------------------------------------------------------
 .../CircuitBreakerLoadBalancer.java             | 55 ++++++++++++++----
 .../CircuitBreakerLoadBalancerTest.java         | 60 +++++++++++++++-----
 2 files changed, 91 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/799b45df/camel-core/src/main/java/org/apache/camel/processor/loadbalancer/CircuitBreakerLoadBalancer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/loadbalancer/CircuitBreakerLoadBalancer.java
b/camel-core/src/main/java/org/apache/camel/processor/loadbalancer/CircuitBreakerLoadBalancer.java
index b8e23b4..3e84e6e 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/loadbalancer/CircuitBreakerLoadBalancer.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/loadbalancer/CircuitBreakerLoadBalancer.java
@@ -107,6 +107,13 @@ public class CircuitBreakerLoadBalancer extends LoadBalancerSupport implements
T
 
         if (failures.get() >= threshold && System.currentTimeMillis() - lastFailure
< halfOpenAfter) {
             exchange.setException(new RejectedExecutionException("CircuitBreaker Open: failures:
" + failures + ", lastFailure: " + lastFailure));
+            /*
+             * If the circuit opens, we have to prevent the execution of any processor.
+             * The failures count can be set to 0.
+             */
+            failures.set(0);
+            callback.done(true);
+            return true;
         }
         Processor processor = getProcessors().get(0);
         if (processor == null) {
@@ -114,18 +121,20 @@ public class CircuitBreakerLoadBalancer extends LoadBalancerSupport
implements T
         }
 
         AsyncProcessor albp = AsyncProcessorConverterHelper.convert(processor);
-        boolean sync = albp.process(exchange, callback);
-
-        boolean failed = hasFailed(exchange);
-
-        if (!failed) {
-            failures.set(0);
+        // Added a callback for processing the exchange in the callback
+        boolean sync = albp.process(exchange, new CircuitBreakerCallback(exchange, callback));
+        
+        // We need to check the exception here as albp is use sync call  
+        if (sync) {
+            boolean failed = hasFailed(exchange);
+            if (!failed) {
+                failures.set(0);
+            } else {
+                failures.incrementAndGet();
+                lastFailure = System.currentTimeMillis();
+            }
         } else {
-            failures.incrementAndGet();
-            lastFailure = System.currentTimeMillis();
-        }
-
-        if (!sync) {
+            // CircuitBreakerCallback can take care of failure check of the exchange
             log.trace("Processing exchangeId: {} is continued being processed asynchronously",
exchange.getExchangeId());
             return false;
         }
@@ -142,4 +151,28 @@ public class CircuitBreakerLoadBalancer extends LoadBalancerSupport implements
T
     public String getTraceLabel() {
         return "circuitbreaker";
     }
+    
+    class CircuitBreakerCallback implements AsyncCallback {
+        private final AsyncCallback callback;
+        private final Exchange exchange;
+        CircuitBreakerCallback(Exchange exchange, AsyncCallback callback) {
+            this.callback = callback;
+            this.exchange = exchange;
+        }
+
+        @Override
+        public void done(boolean doneSync) {
+            if (!doneSync) {
+                boolean failed = hasFailed(exchange);
+                if (!failed) {
+                    failures.set(0);
+                } else {
+                    failures.incrementAndGet();
+                    lastFailure = System.currentTimeMillis();
+                }
+            }
+            callback.done(doneSync);
+        }
+        
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/799b45df/camel-core/src/test/java/org/apache/camel/processor/CircuitBreakerLoadBalancerTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/CircuitBreakerLoadBalancerTest.java
b/camel-core/src/test/java/org/apache/camel/processor/CircuitBreakerLoadBalancerTest.java
index 1229748..a6e489d 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/CircuitBreakerLoadBalancerTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/CircuitBreakerLoadBalancerTest.java
@@ -23,6 +23,7 @@ import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.component.mock.MockEndpoint;
+
 import static org.apache.camel.component.mock.MockEndpoint.expectsMessageCount;
 
 public class CircuitBreakerLoadBalancerTest extends ContextTestSupport {
@@ -46,19 +47,42 @@ public class CircuitBreakerLoadBalancerTest extends ContextTestSupport
{
                 from("direct:start").loadBalance()
                     .circuitBreaker(2, 1000L, MyCustomException.class)
                         .to("mock:result");
+                from("direct:start-async").loadBalance()
+                .circuitBreaker(2, 1000L, MyCustomException.class)
+                    .threads(1).to("mock:result");
             }
         };
     }
 
-    public void testClosedCircuitPassesMessages() throws Exception {
+    public void testClosedCircuitPassesMessagesSync() throws Exception {
+        String endpoint = "direct:start";
+        closedCircuitPassesMessages(endpoint);
+    }
+    
+    public void testClosedCircuitPassesMessagesAsync() throws Exception {
+        String endpoint = "direct:start-async";
+        closedCircuitPassesMessages(endpoint);
+    }
+
+    private void closedCircuitPassesMessages(String endpoint) throws InterruptedException,
Exception {
         expectsMessageCount(3, result);
-        sendMessage("direct:start", "message one");
-        sendMessage("direct:start", "message two");
-        sendMessage("direct:start", "message three");
+        sendMessage(endpoint, "message one");
+        sendMessage(endpoint, "message two");
+        sendMessage(endpoint, "message three");
         assertMockEndpointsSatisfied();
     }
 
-    public void testFailedMessagesOpenCircuitToPreventMessageThree() throws Exception {
+    public void testFailedMessagesOpenCircuitToPreventMessageThreeSync() throws Exception
{
+        String endpoint = "direct:start";
+        failedMessagesOpenCircuitToPreventMessageThree(endpoint);
+    }
+    
+    public void testFailedMessagesOpenCircuitToPreventMessageThreeAsync() throws Exception
{
+        String endpoint = "direct:start-async";
+        failedMessagesOpenCircuitToPreventMessageThree(endpoint);
+    }
+
+    private void failedMessagesOpenCircuitToPreventMessageThree(String endpoint) throws InterruptedException,
Exception {
         expectsMessageCount(2, result);
 
         result.whenAnyExchangeReceived(new Processor() {
@@ -68,9 +92,9 @@ public class CircuitBreakerLoadBalancerTest extends ContextTestSupport {
             }
         });
 
-        Exchange exchangeOne = sendMessage("direct:start", "message one");
-        Exchange exchangeTwo = sendMessage("direct:start", "message two");
-        Exchange exchangeThree = sendMessage("direct:start", "message three");
+        Exchange exchangeOne = sendMessage(endpoint, "message one");
+        Exchange exchangeTwo = sendMessage(endpoint, "message two");
+        Exchange exchangeThree = sendMessage(endpoint, "message three");
         assertMockEndpointsSatisfied();
 
         assertTrue(exchangeOne.getException() instanceof MyCustomException);
@@ -78,7 +102,17 @@ public class CircuitBreakerLoadBalancerTest extends ContextTestSupport
{
         assertTrue(exchangeThree.getException() instanceof RejectedExecutionException);
     }
 
-    public void testHalfOpenCircuitClosesAfterTimeout() throws Exception {
+    public void testHalfOpenCircuitClosesAfterTimeoutSync() throws Exception {
+        String endpoint = "direct:start";
+        halfOpenCircuitClosesAfterTimeout(endpoint);
+    }
+    
+    public void testHalfOpenCircuitClosesAfterTimeoutAsync() throws Exception {
+        String endpoint = "direct:start-async";
+        halfOpenCircuitClosesAfterTimeout(endpoint);
+    }
+
+    private void halfOpenCircuitClosesAfterTimeout(String endpoint) throws InterruptedException,
Exception {
         expectsMessageCount(2, result);
         result.whenAnyExchangeReceived(new Processor() {
             @Override
@@ -87,16 +121,16 @@ public class CircuitBreakerLoadBalancerTest extends ContextTestSupport
{
             }
         });
 
-        sendMessage("direct:start", "message one");
-        sendMessage("direct:start", "message two");
-        sendMessage("direct:start", "message three");
+        sendMessage(endpoint, "message one");
+        sendMessage(endpoint, "message two");
+        sendMessage(endpoint, "message three");
         assertMockEndpointsSatisfied();
 
         result.reset();
         expectsMessageCount(1, result);
 
         Thread.sleep(1000);
-        sendMessage("direct:start", "message four");
+        sendMessage(endpoint, "message four");
         assertMockEndpointsSatisfied();
     }
 


Mime
View raw message