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
The following commit(s) were added to refs/heads/main by this push:
new 79100049c220 CAMEL-24928: camel-core - Throttle EIP with asyncDelayed
should not fail with NullPointerException when no permit is queued
79100049c220 is described below
commit 79100049c220f3b0c94b365f19e413b80638565b
Author: smjain <[email protected]>
AuthorDate: Wed Sep 23 17:05:56 2026 +0530
CAMEL-24928: camel-core - Throttle EIP with asyncDelayed should not fail
with NullPointerException when no permit is queued
With asyncDelayed, the (total requests) throttler schedules an exchange that
has to wait at the delay of the next permit in its queue. When the queue has
no permit at all, which is the case when the maximum requests is 0, or when
the only permits are taken by other exchanges at that moment, peek() returns
null and the exchange failed with a NullPointerException. The other modes
wait instead.
When no permit is queued, try again after one period.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/processor/TotalRequestsThrottler.java | 6 +-
.../ThrottlerAsyncDelayedNoPermitTest.java | 69 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 1 deletion(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
index 2cd000321e1a..a2c1c2121323 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
@@ -184,7 +184,11 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
exchange.setProperty(PROPERTY_EXCHANGE_QUEUED_TIMESTAMP,
System.nanoTime());
}
exchange.setProperty(PROPERTY_EXCHANGE_STATE, State.ASYNC);
- long delay = throttlingState.peek().getDelay(TimeUnit.NANOSECONDS);
+ ThrottlePermit next = throttlingState.peek();
+ // there is no permit in the queue when the rate is 0, or when the
only permits are taken by other
+ // exchanges and not yet returned, so try again after one period
+ long delay = next != null
+ ? next.getDelay(TimeUnit.NANOSECONDS) :
TimeUnit.MILLISECONDS.toNanos(getTimePeriodMillis());
asyncExecutor.schedule(() -> process(exchange, callback), delay,
TimeUnit.NANOSECONDS);
return false;
} catch (final RejectedExecutionException e) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/ThrottlerAsyncDelayedNoPermitTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/ThrottlerAsyncDelayedNoPermitTest.java
new file mode 100644
index 000000000000..05909ca79989
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/ThrottlerAsyncDelayedNoPermitTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.throttle.requests;
+
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.support.ExpressionAdapter;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * With asyncDelayed, an exchange that finds no permit at all in the queue
(the maximum requests is 0) should wait and
+ * try again, instead of failing.
+ */
+public class ThrottlerAsyncDelayedNoPermitTest extends ContextTestSupport {
+
+ private final AtomicInteger evaluations = new AtomicInteger();
+
+ @Test
+ public void testNoPermitAsyncDelayed() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ Future<Exchange> reply = template.asyncSend("direct:start", e ->
e.getMessage().setBody("Hello World"));
+ Exchange out = reply.get(10, TimeUnit.SECONDS);
+
+ assertNull(out.getException());
+ assertMockEndpointsSatisfied();
+ // the maximum requests was evaluated as 0 first, and as 1 when the
exchange tried again
+ assertEquals(2, evaluations.get());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:start")
+ .throttle(new ExpressionAdapter() {
+ @Override
+ public Object evaluate(Exchange exchange) {
+ // paused for the first evaluation, then 1 per
period
+ return evaluations.incrementAndGet() == 1 ? 0
: 1;
+ }
+ }).timePeriodMillis(100).asyncDelayed()
+ .to("mock:result");
+ }
+ };
+ }
+}
