This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 80210ce04b23ac71cb1deb92018617d4d7ee49b6 Author: smjain <[email protected]> AuthorDate: Wed Sep 23 17:34:12 2026 +0530 CAMEL-24922: camel-core - Stop sticky failover by counting the endpoints tried The previous change stopped the wrap around when the index came back to the start index. If the last known good index is out of range, for example after processors were removed from a running load balancer, the index never came back to it and the failover kept wrapping around with maximumFailoverAttempts -1. Count the endpoints tried for the exchange instead, and stop when all of them have been tried once. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../loadbalancer/FailOverLoadBalancer.java | 9 +++-- .../processor/FailoverStickyWrapAroundTest.java | 47 ++++++++++++++++++++++ 2 files changed, 52 insertions(+), 4 deletions(-) diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/FailOverLoadBalancer.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/FailOverLoadBalancer.java index a3ffb283509c..d6be8d9a4ea4 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/FailOverLoadBalancer.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/FailOverLoadBalancer.java @@ -181,8 +181,9 @@ public class FailOverLoadBalancer extends LoadBalancerSupport implements Traceab final AsyncCallback callback; final AsyncProcessor[] processors; int index; - int start; int attempts; + // number of endpoints tried for this exchange + int tried; // use a copy of the original exchange before failover to avoid populating side effects // directly into the original exchange Exchange copy; @@ -199,7 +200,6 @@ public class FailOverLoadBalancer extends LoadBalancerSupport implements Traceab } else if (isRoundRobin()) { index = counter.updateAndGet(x -> ++x < processors.length ? x : 0); } - start = index; LOG.trace("Failover starting with endpoint index {}", index); } @@ -247,7 +247,7 @@ public class FailOverLoadBalancer extends LoadBalancerSupport implements Traceab LOG.trace("Failover is round robin enabled and therefore starting from the first endpoint"); index = 0; counter.set(0); - } else if (isSticky() && start > 0) { + } else if (isSticky() && tried < processors.length) { // sticky mode started from the last known good endpoint, so the endpoints // before it have not been tried yet LOG.trace("Failover is sticky enabled and therefore continuing from the first endpoint"); @@ -261,7 +261,7 @@ public class FailOverLoadBalancer extends LoadBalancerSupport implements Traceab } } - if (copy != null && isSticky() && !isRoundRobin() && index == start) { + if (isSticky() && !isRoundRobin() && tried >= processors.length) { // sticky mode (without round robin) has tried all endpoints once LOG.trace("Breaking out of failover as all endpoints have been tried"); ExchangeHelper.copyResults(exchange, copy); @@ -270,6 +270,7 @@ public class FailOverLoadBalancer extends LoadBalancerSupport implements Traceab } // try again but copy original exchange before we failover + tried++; copy = prepareExchangeForFailover(exchange); AsyncProcessor processor = processors[index]; diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/FailoverStickyWrapAroundTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/FailoverStickyWrapAroundTest.java index aee129605896..f63ca81a5d97 100644 --- a/core/camel-core/src/test/java/org/apache/camel/processor/FailoverStickyWrapAroundTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/processor/FailoverStickyWrapAroundTest.java @@ -18,12 +18,17 @@ package org.apache.camel.processor; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import org.apache.camel.CamelExecutionException; import org.apache.camel.ContextTestSupport; import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.processor.loadbalancer.FailOverLoadBalancer; import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; /** @@ -32,6 +37,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; public class FailoverStickyWrapAroundTest extends ContextTestSupport { private final Set<String> down = ConcurrentHashMap.newKeySet(); + private final FailOverLoadBalancer dynamic = createStickyFailOver(); @Test public void testFailoverStickyContinuesFromFirstEndpoint() throws Exception { @@ -117,6 +123,43 @@ public class FailoverStickyWrapAroundTest extends ContextTestSupport { assertMockEndpointsSatisfied(); } + @Test + public void testFailoverStickyWhenLastGoodEndpointWasRemoved() throws Exception { + // a and b are down, so c becomes the last known good endpoint + down.add("a"); + down.add("b"); + template.sendBody("direct:dynamic", "Hello World"); + assertEquals(2, dynamic.getLastGoodIndex()); + + // c is removed while the route is running, so the last known good index is now out of range + dynamic.removeProcessor(dynamic.getProcessors().get(2)); + + // all remaining endpoints are down: each is tried once and the exchange fails (it must not loop) + resetMocks(); + getMockEndpoint("mock:a").expectedMessageCount(1); + getMockEndpoint("mock:b").expectedMessageCount(1); + getMockEndpoint("mock:c").expectedMessageCount(0); + Future<Object> future = template.asyncRequestBody("direct:dynamic", "Bye World"); + assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); + assertMockEndpointsSatisfied(); + + // b is up again + resetMocks(); + down.remove("b"); + getMockEndpoint("mock:a").expectedMessageCount(1); + getMockEndpoint("mock:b").expectedBodiesReceived("Hi World"); + template.sendBody("direct:dynamic", "Hi World"); + assertMockEndpointsSatisfied(); + } + + private static FailOverLoadBalancer createStickyFailOver() { + FailOverLoadBalancer answer = new FailOverLoadBalancer(); + answer.setMaximumFailoverAttempts(-1); + answer.setRoundRobin(false); + answer.setSticky(true); + return answer; + } + @Override protected RouteBuilder createRouteBuilder() { return new RouteBuilder() { @@ -130,6 +173,10 @@ public class FailoverStickyWrapAroundTest extends ContextTestSupport { .loadBalance().failover(2, false, false, true) .to("direct:a", "direct:b", "direct:c", "direct:d"); + from("direct:dynamic") + .loadBalance(dynamic) + .to("direct:a", "direct:b", "direct:c"); + from("direct:a").to("mock:a").process(e -> failIfDown("a")); from("direct:b").to("mock:b").process(e -> failIfDown("b")); from("direct:c").to("mock:c").process(e -> failIfDown("c"));
