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 668c9299d846 CAMEL-24927: camel-core - Sample EIP should let the first
exchange through whatever the value of System.nanoTime() is
668c9299d846 is described below
commit 668c9299d84648c7e6e1ff39f5402160216256dc
Author: smjain <[email protected]>
AuthorDate: Wed Sep 23 17:02:16 2026 +0530
CAMEL-24927: camel-core - Sample EIP should let the first exchange through
whatever the value of System.nanoTime() is
Since CAMEL-20267 the SamplingThrottler measures time with
System.nanoTime(),
but it still starts with timeOfLastExchange = 0 and lets an exchange through
when now >= timeOfLastExchange + period. The origin of System.nanoTime() is
arbitrary, and on Linux and macOS it is about the time since the machine was
started. So as long as the machine has been up for less than the sample
period, every exchange was dropped, for example all of them for up to a day
with sample(Duration.ofDays(1)) on a machine started in the morning.
The first exchange is now always let through, and the period is measured
from there.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/processor/SamplingThrottler.java | 5 ++++-
.../apache/camel/processor/SamplingThrottlerTest.java | 16 ++++++++++++++++
2 files changed, 20 insertions(+), 1 deletion(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/SamplingThrottler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/SamplingThrottler.java
index de903397e0fc..c46d7ee12464 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/SamplingThrottler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/SamplingThrottler.java
@@ -53,6 +53,7 @@ public class SamplingThrottler extends BaseProcessorSupport
implements Traceable
private long periodInMillis;
private TimeUnit units;
private long timeOfLastExchange;
+ private boolean sampledOnce;
private final StopProcessor stopper = new StopProcessor();
private final Lock calculationLock = new ReentrantLock();
private final SampleStats sampled = new SampleStats();
@@ -145,7 +146,9 @@ public class SamplingThrottler extends BaseProcessorSupport
implements Traceable
}
} else {
long now = Duration.ofNanos(System.nanoTime()).toMillis();
- if (now >= timeOfLastExchange + periodInMillis) {
+ // System.nanoTime() has an arbitrary origin, so the first
exchange cannot be compared with a time
+ if (!sampledOnce || now >= timeOfLastExchange +
periodInMillis) {
+ sampledOnce = true;
doSend = true;
if (LOG.isTraceEnabled()) {
LOG.trace(sampled.sample());
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SamplingThrottlerTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SamplingThrottlerTest.java
index b98e5e9f9e68..c3851ff94773 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/SamplingThrottlerTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SamplingThrottlerTest.java
@@ -54,6 +54,20 @@ public class SamplingThrottlerTest extends
ContextTestSupport {
validateDroppedExchanges(sentExchanges, mock.getReceivedCounter());
}
+ @Test
+ public void testFirstExchangeIsSampledWithLongPeriod() throws Exception {
+ // the first exchange is sampled whatever the period is, also when it
is longer than the time since the
+ // (arbitrary) origin of System.nanoTime(), which is usually the time
since the machine was started
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("1");
+
+ for (int i = 1; i <= 5; i++) {
+ template.sendBody("direct:sample-long-period", String.valueOf(i));
+ }
+
+ mock.assertIsSatisfied();
+ }
+
@Test
public void testBurstySampling() throws Exception {
NotifyBuilder notify = new NotifyBuilder(context).whenDone(5).create();
@@ -190,6 +204,8 @@ public class SamplingThrottlerTest extends
ContextTestSupport {
from("direct:sample-messageFrequency-via-dsl").sample().sampleMessageFrequency(5).to("mock:result");
+
from("direct:sample-long-period").sample(Duration.ofDays(36500)).to("mock:result");
+
from("direct:sample-placeholder").sample("{{sample.period}}").to("mock:result");
// END SNIPPET: e1
