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 cc6102541d92421dad4d7b72b45d7ccc1a8fe609 Author: smjain <[email protected]> AuthorDate: Wed Sep 23 16:41:38 2026 +0530 CAMEL-24923: camel-core - Validate the distribution ratios of the weighted load balancer A weighted load balancer accepted negative ratios, ratios that are all 0, and ratios whose sum overflows an int. In round robin mode the processor selection then loops forever while holding the lock, and in random mode nextInt throws from process() without calling the callback, so the caller hangs in both cases. An overflowing sum can also send every message to the first endpoint. The ratios are now checked when the load balancer starts, in the same place the number of ratios is already checked. A ratio of 0 for some of the endpoints is still allowed. QueueLoadBalancer also sets an exception thrown while choosing a processor on the exchange and calls the callback, instead of letting it escape process(). Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../catalog/docs/weightedLoadBalancer-eip.adoc | 4 + .../eips/pages/weightedLoadBalancer-eip.adoc | 4 + .../processor/loadbalancer/QueueLoadBalancer.java | 9 +- .../loadbalancer/WeightedLoadBalancer.java | 18 ++++ .../WeightedLoadBalanceInvalidRatioTest.java | 107 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 7 ++ 6 files changed, 148 insertions(+), 1 deletion(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/weightedLoadBalancer-eip.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/weightedLoadBalancer-eip.adoc index 94d0f8023508..4e8422397d96 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/weightedLoadBalancer-eip.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/weightedLoadBalancer-eip.adoc @@ -28,6 +28,10 @@ In this example, we want to send the most message to the first endpoint, then th The distribution ratio is `7 = 4 + 2 + 1`. This means that for every seventh message then 4 goes to the first, 2 for the second, and 1 for the last. +Each ratio must be zero or a positive number, at least one ratio must be positive, and the sum of the ratios +must not be greater than `2147483647`. Otherwise, the route fails to start. +An endpoint with ratio `0` does not receive any messages. + [tabs] ==== diff --git a/core/camel-core-engine/src/main/docs/modules/eips/pages/weightedLoadBalancer-eip.adoc b/core/camel-core-engine/src/main/docs/modules/eips/pages/weightedLoadBalancer-eip.adoc index 94d0f8023508..4e8422397d96 100644 --- a/core/camel-core-engine/src/main/docs/modules/eips/pages/weightedLoadBalancer-eip.adoc +++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/weightedLoadBalancer-eip.adoc @@ -28,6 +28,10 @@ In this example, we want to send the most message to the first endpoint, then th The distribution ratio is `7 = 4 + 2 + 1`. This means that for every seventh message then 4 goes to the first, 2 for the second, and 1 for the last. +Each ratio must be zero or a positive number, at least one ratio must be positive, and the sum of the ratios +must not be greater than `2147483647`. Otherwise, the route fails to start. +An endpoint with ratio `0` does not receive any messages. + [tabs] ==== diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/QueueLoadBalancer.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/QueueLoadBalancer.java index 710df651a577..64d29bb702de 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/QueueLoadBalancer.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/QueueLoadBalancer.java @@ -30,7 +30,14 @@ public abstract class QueueLoadBalancer extends LoadBalancerSupport { public boolean process(final Exchange exchange, final AsyncCallback callback) { AsyncProcessor[] list = doGetProcessors(); if (list.length > 0) { - AsyncProcessor processor = chooseProcessor(list, exchange); + AsyncProcessor processor; + try { + processor = chooseProcessor(list, exchange); + } catch (Exception e) { + exchange.setException(e); + callback.done(true); + return true; + } if (processor == null) { Exception e = new IllegalStateException("No processors could be chosen to process " + exchange); exchange.setException(e); diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/WeightedLoadBalancer.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/WeightedLoadBalancer.java index 581abfcd9e9a..7b37670e403f 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/WeightedLoadBalancer.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/loadbalancer/WeightedLoadBalancer.java @@ -46,6 +46,24 @@ public abstract class WeightedLoadBalancer extends QueueLoadBalancer { "Loadbalacing with " + getProcessors().size() + " should match number of distributions " + ratios.size()); } + // a ratio that is negative, or ratios that are all zero or add up to more than an int can hold, + // would make the processor selection loop forever or fail on every exchange + long sum = 0; + for (DistributionRatio ratio : ratios) { + int weight = ratio.getDistributionWeight(); + if (weight < 0) { + throw new IllegalArgumentException( + "Distribution ratio must be zero or a positive number, was: " + weight); + } + sum += weight; + } + if (sum == 0) { + throw new IllegalArgumentException("At least one distribution ratio must be a positive number"); + } + if (sum > Integer.MAX_VALUE) { + throw new IllegalArgumentException( + "The sum of the distribution ratios must not be greater than " + Integer.MAX_VALUE + ", was: " + sum); + } } protected void decrementSum() { diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/WeightedLoadBalanceInvalidRatioTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/WeightedLoadBalanceInvalidRatioTest.java new file mode 100644 index 000000000000..9caad4883537 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/WeightedLoadBalanceInvalidRatioTest.java @@ -0,0 +1,107 @@ +/* + * 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; + +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.camel.AsyncProcessor; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.processor.loadbalancer.QueueLoadBalancer; +import org.apache.camel.support.DefaultExchange; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class WeightedLoadBalanceInvalidRatioTest extends ContextTestSupport { + + @Override + public boolean isUseRouteBuilder() { + return false; + } + + @ParameterizedTest + @CsvSource(delimiter = ';', value = { + "true;0,0;At least one distribution ratio must be a positive number", + "false;0,0;At least one distribution ratio must be a positive number", + "true;1,-1;Distribution ratio must be zero or a positive number, was: -1", + "false;1,-1;Distribution ratio must be zero or a positive number, was: -1", + "true;2147483647,1;The sum of the distribution ratios must not be greater than 2147483647, was: 2147483648", + "false;2147483647,1;The sum of the distribution ratios must not be greater than 2147483647, was: 2147483648", + "true;2147483647,2147483647,3,0;The sum of the distribution ratios must not be greater than 2147483647, was: 4294967297" }) + public void testInvalidRatiosRejectedOnStart(boolean roundRobin, String ratios, String message) throws Exception { + // one endpoint per ratio, so the number of ratios is valid + String[] uris = new String[ratios.split(",").length]; + for (int i = 0; i < uris.length; i++) { + uris[i] = "mock:" + i; + } + context.addRoutes(new RouteBuilder() { + public void configure() { + from("direct:start").loadBalance().weighted(roundRobin, ratios).to(uris); + } + }); + + Exception e = assertThrows(Exception.class, () -> context.start()); + IllegalArgumentException iae = assertIsInstanceOf(IllegalArgumentException.class, e.getCause().getCause()); + assertEquals(message, iae.getMessage()); + } + + @Test + public void testZeroRatioForSomeProcessorsIsAllowed() throws Exception { + context.addRoutes(new RouteBuilder() { + public void configure() { + from("direct:start").loadBalance().weighted(true, "0,2,1").to("mock:x", "mock:y", "mock:z"); + } + }); + context.start(); + + getMockEndpoint("mock:x").expectedMessageCount(0); + getMockEndpoint("mock:y").expectedMessageCount(4); + getMockEndpoint("mock:z").expectedMessageCount(2); + + for (int i = 0; i < 6; i++) { + template.sendBody("direct:start", "Hello " + i); + } + + assertMockEndpointsSatisfied(); + } + + @Test + public void testExceptionWhenChoosingProcessorIsSetOnExchange() throws Exception { + IllegalStateException cause = new IllegalStateException("Cannot choose"); + QueueLoadBalancer lb = new QueueLoadBalancer() { + @Override + protected AsyncProcessor chooseProcessor(AsyncProcessor[] processors, Exchange exchange) { + throw cause; + } + }; + lb.addProcessor(new SendProcessor(context.getEndpoint("mock:x"))); + + Exchange exchange = new DefaultExchange(context); + AtomicBoolean done = new AtomicBoolean(); + lb.process(exchange, doneSync -> done.set(true)); + + assertTrue(done.get(), "The callback should be called"); + assertSame(cause, exchange.getException()); + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc index e4ec6f31e52b..cfe47db7bc9a 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc @@ -73,6 +73,13 @@ Prior to Camel 4.23 the property was only set when there was no fallback and was so a fallback that tested it for `null` must now test for `true` or `false` instead. `CamelCircuitBreakerResponseShortCircuited` is unchanged and remains `true` whenever the fallback runs, whatever the cause. +=== Weighted Load Balancer EIP + +The distribution ratios of the weighted load balancer are now validated when the route starts. A negative ratio, +ratios that are all `0`, or ratios whose sum is greater than `2147483647` now fail the route at startup with an +`IllegalArgumentException`. Previously such a route started, but sending to it could hang the caller, spin a CPU, +or send every message to the same endpoint. + === Context reload now re-applies placeholder based component options When a context reload is triggered, for example by one of the vault components detecting that a secret was rotated,
