This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 c536c99caaaa CAMEL-25911: camel-test-dst - stage 2: an internal
TimeSource in camel-util for core waits, task timing and throttling (#28382)
c536c99caaaa is described below
commit c536c99caaaa81a6e309b20b30881a3c9ec60dab
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 25 15:14:02 2026 +0530
CAMEL-25911: camel-test-dst - stage 2: an internal TimeSource in camel-util
for core waits, task timing and throttling (#28382)
Core code that waits, times tasks or throttles (redelivery, delayer,
throttlers, ForegroundTask, the shutdown wait loop, StopWatch, MonotonicClock)
now reads time and sleeps through an internal TimeSource in camel-util. The
default provider makes the same JDK calls as before; camel-test-dst installs a
virtual-time provider so these waits run on the seeded scheduler. The throttler
TRACE "Queued for {}ms" lines now log milliseconds, and test-dst.adoc lists the
waits that stay outside vi [...]
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/test-dst.adoc | 77 ++++++---
.../aws2/sqs/dst/SqsRedeliveryDstTest.java | 37 +++++
.../camel/component/aws2/sqs/dst/SqsWorld.java | 58 ++++++-
.../camel-test-dst/src/main/docs/test-dst.adoc | 77 ++++++---
.../main/java/org/apache/camel/test/dst/Dst.java | 21 ++-
.../java/org/apache/camel/test/dst/DstClock.java | 53 ++++++
.../org/apache/camel/test/dst/DstExtension.java | 31 ++--
.../org/apache/camel/test/dst/DstInstrument.java | 19 ++-
.../apache/camel/test/dst/DstInstrumentation.java | 16 +-
.../apache/camel/test/dst/DstAlsoRewriteTest.java | 34 ++--
.../apache/camel/test/dst/DstCoreWaitsTest.java | 71 ++++++--
.../apache/camel/test/dst/DstTimeSourceTest.java | 179 +++++++++++++++++++++
.../camel/impl/engine/DefaultShutdownStrategy.java | 3 +-
.../processor/ConcurrentRequestsThrottler.java | 11 +-
.../camel/processor/DelayProcessorSupport.java | 5 +-
.../apache/camel/processor/MulticastProcessor.java | 3 +-
.../apache/camel/processor/SamplingThrottler.java | 5 +-
.../camel/processor/TotalRequestsThrottler.java | 13 +-
.../aggregate/OptimisticLockRetryPolicy.java | 4 +-
.../errorhandler/RedeliveryErrorHandler.java | 3 +-
.../processor/errorhandler/RedeliveryPolicy.java | 3 +-
.../apache/camel/support/DefaultTimeoutMap.java | 3 +-
.../org/apache/camel/support/MonotonicClock.java | 7 +-
.../startup/DefaultStartupConditionStrategy.java | 3 +-
.../apache/camel/support/task/AbstractTask.java | 4 +-
.../apache/camel/support/task/BackgroundTask.java | 5 +-
.../apache/camel/support/task/ForegroundTask.java | 13 +-
.../task/budget/IterationBoundedBudget.java | 7 +-
.../support/task/budget/TimeBoundedBudget.java | 6 +-
.../throttling/ThrottlingExceptionRoutePolicy.java | 13 +-
.../main/java/org/apache/camel/util/StopWatch.java | 10 +-
.../java/org/apache/camel/util/TimeSource.java | 172 ++++++++++++++++++++
.../camel/util/backoff/BackOffTimerTask.java | 3 +-
.../java/org/apache/camel/util/TimeSourceTest.java | 162 +++++++++++++++++++
34 files changed, 986 insertions(+), 145 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
index 0d13aa8d264a..16bf1c96d140 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
@@ -16,13 +16,16 @@ with the same step trace.
* Every thread pool created through the `ExecutorServiceManager` is backed by
one event loop on the test thread
(`DstThreadPoolFactory`). Tasks run one at a time; when several can run, the
seed picks one.
* Time is virtual: it jumps to the next due task. The time reads, sleeps and
blocking waits of the code under test are
- rewritten at test time (`@DstInstrument`), so "wait 30 seconds" costs
nothing.
+ rewritten at test time (`@DstInstrument`), and the Camel core code that
waits, times tasks or throttles reads the
+ virtual clock through its internal time source, so "wait 30 seconds" costs
nothing.
* A fake of the external system (a queue, a checkpoint store, a web API)
injects faults chosen by the same seed.
* Invariants are checked at the end of each run.
* `@DstTest(seeds = 1000)` runs a test method once per seed. A failing seed is
printed with its trace and is replayed
exactly with `-Ddst.seed=4711` or `@DstSeed(4711)`.
-This is a test tool. Nothing of it runs in production, and it changes nothing
in Camel.
+This is a test tool. Nothing of it runs in production. Camel core has one
internal hook for it, `TimeSource` in
+camel-util: with its default provider it reads `System` and sleeps with
`Thread.sleep`, and only an open run replaces
+the provider.
== A test
@@ -54,6 +57,8 @@ These worlds are in the Camel code base and are the examples
to copy:
|Component |Test |Regression test for
|camel-aws2-sqs |`SqsVisibilityDstTest` (the visibility timeout extender)
|CAMEL-25700
+|camel-aws2-sqs |`SqsRedeliveryDstTest` (the extender while Camel's error
handler waits to redeliver, in virtual
+time) |-
|camel-azure-eventhubs |`EventHubsCheckpointDstTest` (checkpointing by batch
size and timeout) |CAMEL-25683
|camel-pulsar |`PulsarConsumerDstTest` (subscribe failures, broker and route
restarts while the route starts and
consumes) |CAMEL-25869
@@ -78,13 +83,25 @@ virtual time. A wait that no task of the run can end fails
the seed at once and
|Timed waits on tasks |`Future.get(timeout)` on a task of the run and
`awaitTermination` time out in virtual time.
|Parallel streams |`parallelStream()` of a `java.util` collection is rewritten
to `stream()`: the elements are handled
in order by the calling task, not on the common fork-join pool.
+|Camel core |The code of Camel core that waits, times tasks or throttles reads
the clock and sleeps through
+`org.apache.camel.util.TimeSource`, an internal class of camel-util. A run
installs its virtual clock there (`Dst`
+does it when the run is created) and puts the previous provider back when the
run is closed, so no rewriting is needed:
+the synchronous redelivery (`RedeliveryPolicy`, `RedeliveryErrorHandler`),
`ForegroundTask`, `BackgroundTask.schedule`,
+the budgets of both tasks, the wait loop of `DefaultShutdownStrategy`, the
synchronous delayer, `MulticastProcessor`,
+`OptimisticLockRetryPolicy`, `DefaultStartupConditionStrategy`, `StopWatch`,
`MonotonicClock`, `BackOffTimerTask`,
+`DefaultTimeoutMap`, the clocks of the throttlers and the counters of
`ThrottlingExceptionRoutePolicy`. A sleep there is
+a virtual wait on the loop thread, traced with its call site. Three waits of
these classes are not virtual (see below):
+the permit queue of `TotalRequestsThrottler`, the `Semaphore` of
`ConcurrentRequestsThrottler` and the half-open timer
+of `ThrottlingExceptionRoutePolicy`. The sleep of the route `delayer` option
is not virtual either (see below).
|===
-`@DstInstrument(alsoRewrite = ...)` applies the same rewriting to classes
outside the code under test, for example
-`org.apache.camel.util.StopWatch`, `org.apache.camel.support.MonotonicClock`,
`org.apache.camel.support.task`
-(`ForegroundTask`), `org.apache.camel.processor.errorhandler` (synchronous
redelivery) or
-`org.apache.camel.impl.engine.DefaultShutdownStrategy` (its wait for in-flight
exchanges). This is a stop-gap: Camel
-core reads `System` directly, and a time source in core that the harness can
replace needs a dev@ discussion first.
+Event timestamps, the clock of an exchange and the direct clock reads of the
debugger, the tracer, management and the
+developer consoles stay on `System`. The durations they measure with
`StopWatch` or `MonotonicClock` follow `TimeSource`
+(with the default provider this is the same `System.nanoTime()`), and so does
the created time of `MonotonicClock`: the
+message history times (`MessageHistory.getTime()`) are virtual in a DST run.
`@DstInstrument(alsoRewrite = ...)` applies the
+rewriting of the code under test to classes outside it, for example
`org.apache.camel.support.ResetableClock` (the
+clock of an exchange) or a library the component calls. It is not needed for
the core classes listed above, except for
+the permit queue of `TotalRequestsThrottler` (see below).
=== Escapes and strict mode
@@ -93,15 +110,34 @@ Before the first seed, the harness scans the packages
under test and writes ever
cannot run the loop for (`Object.wait`, `LockSupport.park`, `Mono.block`),
wall clock reads it does not rewrite.
Strict mode is the default: when a seed reaches one of these at runtime, or a
thread outside the harness submits a task
-to the loop, reads the virtual clock or sleeps, the seed fails with the escape
as its cause. A clock read from such a
+to the loop, reads the virtual clock or sleeps, the seed fails with the escape
as its cause. This includes Camel core:
+a `StopWatch` started on such a thread reads the virtual clock through
`TimeSource`. A clock read from such a
thread (a Reactor or Netty thread, for example) still gets the virtual time
and its sleep still takes real time, so
that thread mixes the two; this is what strict mode is there to make visible.
`@DstInstrument(allowEscapes = "Foo.bar")` allows an escape
that cannot change the result (say why in a comment); `@DstInstrument(strict =
false)` or `-Ddst.strict=false` only
records escapes in `target/dst/<Test>.summary.txt`.
-A wait in code that is not rewritten (a latch in Camel core, for example)
blocks the loop thread for real. A watchdog
-interrupts the loop thread after 10 s of real time without progress
(`-Ddst.watchdogMillis`) and the seed fails with the
-stack where it was blocked.
+A wait in code that is not rewritten blocks the loop thread for real. A
watchdog interrupts the loop thread after 10 s
+of real time without progress (`-Ddst.watchdogMillis`) and the seed fails with
the stack where it was blocked. In Camel
+core:
+
+* `BackgroundTask.run` waits on a latch until its scheduled attempts are done,
so it blocks the loop thread.
+ `BackgroundTask.schedule` does not wait, and its attempts run in virtual
time.
+* `TotalRequestsThrottler` waits for a permit on its `DelayQueue`. The delay
of a permit follows the virtual clock,
+ which does not move while the loop thread waits, so a seed that has to wait
for a permit always ends on the watchdog.
+ `asyncDelayed()` does not avoid it: a delayed exchange that finds its permit
taken by another one waits on the same
+ queue. Add `org.apache.camel.processor.TotalRequestsThrottler` to
`@DstInstrument(alsoRewrite)`: the permit wait is
+ then a virtual wait.
+* `ConcurrentRequestsThrottler` blocks on its `Semaphore` once
`maxConcurrentRequests` exchanges are in flight. No task
+ of the run releases a permit while the loop thread waits, so the seed ends
on the watchdog. `alsoRewrite` does not
+ help here (the call goes through a `Semaphore` subclass): keep
`maxConcurrentRequests` above the number of exchanges
+ the world has in flight.
+* The route `delayer` option (`CamelInternalProcessor.DelayerAdvice`) calls
`Thread.sleep` directly, not through
+ `TimeSource`: the delay takes real time on the loop thread and the virtual
clock does not move. Use the `delay` EIP
+ in a world instead.
+* The half-open check of `ThrottlingExceptionRoutePolicy` runs on a
`java.util.Timer` thread, outside the harness. The
+ timer fires after `halfOpenAfter` of real time (30 s by default, longer than
the watchdog) and its clock read is an
+ escape, so a world does not see the circuit half-open in virtual time.
=== The one-stack model
@@ -404,12 +440,15 @@ seed of the in-JVM replay test of the Event Hubs world,
default 1).
* Code on its own threads or on a third-party scheduler is outside the
harness; strict mode makes reaching it fail the
seed.
* One run per JVM at a time: the active run, and so the virtual clock, is a
static field, and every rewritten clock read
- in the JVM sees it. A world must therefore not go into a module that runs
its tests in parallel
- (`camel.surefire.parallel=true`, as camel-jms and camel-sql do): the other
tests in that surefire fork would read the
- virtual clock while a run is active. The harness enforces this: a second run
while one is open fails, and a DST test
- refuses to start when JUnit runs tests in parallel. A local world in such a
module (camel-sjms, for example) can
- still be run alone, with `-Dtest=<the world>
-Dcamel.surefire.parallel=false`, but it cannot be committed there:
- the module's own build would fail it.
+ in the JVM sees it, as does Camel core through `TimeSource` while the run is
open. A world must therefore not go into
+ a module that runs its tests in parallel (`camel.surefire.parallel=true`, as
camel-jms and camel-sql do): the other
+ tests in that surefire fork would read the virtual clock while a run is
active. For the same reason, a
+ `CamelContext` left running in the same fork by an earlier non-DST test
would read the virtual time too and could
+ fail a strict seed; stop it in that test. The harness enforces this: a second
+ run while one is open fails, and a DST test refuses to start when JUnit runs
tests in parallel. A local world in
+ such a module (camel-sjms, for example) can still be run alone, with
+ `-Dtest=<the world> -Dcamel.surefire.parallel=false`, but it cannot be
committed there: the module's own build would
+ fail it.
* Locks do not exclude: every task runs on the test thread, so a
`ReentrantLock` or `synchronized` block held by a
blocked task is entered again by the tasks nested above it. A critical
section is atomic only because the fake does
not block inside it, and races between two code paths that each hold a lock
cannot be explored.
@@ -427,7 +466,7 @@ seed of the in-JVM replay test of the Event Hubs world,
default 1).
strict mode
|`DstThreadPoolFactory`, `DstExecutor`, `DstTask` |Camel's pools and
schedulers backed by the loop
|`DstInstrumentation`, `DstClock`, `DstWaits` |the byte code rewrite of the
code under test: time reads, sleeps and
-blocking waits go to the loop
+blocking waits go to the loop; `DstClock` is also the `TimeSource` provider of
Camel core during a run
|`DstEscapes`, `DstWatchdog` |the escape scan, runtime escapes (strict mode),
and the watchdog for waits in code that is
not rewritten
|`DstLedger`, `Invariant`, `Invariants` |what happened to each message, and
reusable invariants
@@ -446,7 +485,7 @@ not rewritten
|`dst.seed` |comma separated seeds to replay; their full traces are written |-
|`dst.traceDir` |where the reports and traces go |`target/dst`
|`dst.strict` |strict mode |`@DstInstrument(strict)`, true
-|`dst.alsoRewrite` |comma separated classes or packages to rewrite too
(stop-gap) |-
+|`dst.alsoRewrite` |comma separated classes or packages outside the code under
test to rewrite too |-
|`dst.watchdogMillis` |real time without progress after which a waiting loop
thread fails the seed |10000
|===
diff --git
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsRedeliveryDstTest.java
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsRedeliveryDstTest.java
new file mode 100644
index 000000000000..c0fef3d9415d
--- /dev/null
+++
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsRedeliveryDstTest.java
@@ -0,0 +1,37 @@
+/*
+ * 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.component.aws2.sqs.dst;
+
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstInstrument;
+import org.apache.camel.test.dst.DstTest;
+
+/**
+ * The SQS world of {@link SqsVisibilityDstTest} with synchronous redelivery:
a failed message is retried by Camel's
+ * error handler after a redelivery delay longer than the visibility timeout,
so the visibility timeout extender has to
+ * keep the message invisible while the error handler waits. The wait is a
sleep in Camel core, which runs in virtual
+ * time through Camel's internal time source; without it each wait would block
the test thread for real (and the seed
+ * would fail on the watchdog or on the invariant that the redelivery delay
passed).
+ */
+@DstInstrument("org.apache.camel.component.aws2.sqs")
+class SqsRedeliveryDstTest {
+
+ @DstTest(seeds = 500)
+ void noMessageIsDeliveredAgainWhileTheErrorHandlerWaitsToRetryIt(Dst dst)
throws Exception {
+ new SqsWorld(dst, true).run();
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsWorld.java
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsWorld.java
index 56873b7dc080..64787fc96ef1 100644
---
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsWorld.java
+++
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsWorld.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.aws2.sqs.Sqs2Constants;
import org.apache.camel.impl.DefaultCamelContext;
@@ -42,10 +43,15 @@ import org.apache.camel.test.dst.Invariants;
* The property under test: while the consumer holds a message (from the
receive until the route has finished with it),
* SQS must not deliver it again, because the extender keeps it invisible. The
only exception is a message whose
* visibility change was lost by a fault of the fake.
+ * <p>
+ * With synchronous redelivery ({@link #SqsWorld(Dst, boolean)}) a failed
message is retried by Camel's error handler
+ * after a redelivery delay longer than the visibility timeout, on the
consumer's thread: the consumer holds the message
+ * during the wait, and the wait of the error handler (a sleep in Camel core)
runs in virtual time.
*/
final class SqsWorld {
static final String ROUTE = "sqs";
+ static final int MAX_REDELIVERIES = 2;
private final Dst dst;
private final DstLedger ledger;
@@ -54,6 +60,10 @@ final class SqsWorld {
private final int maxMessagesPerPoll;
private final int concurrentConsumers;
private final long delay;
+ private final boolean redelivery;
+ private final long redeliveryDelay;
+ private final Map<String, Long> failedAt = new HashMap<>();
+ private final List<String> redeliveredTooEarly = new ArrayList<>();
private final Map<String, Integer> held = new HashMap<>();
private final Set<String> extensionFaulted = new HashSet<>();
private final List<String> redeliveredWhileHeld = new ArrayList<>();
@@ -61,7 +71,16 @@ final class SqsWorld {
private boolean faults = true;
SqsWorld(Dst dst) {
+ this(dst, false);
+ }
+
+ /**
+ * @param redelivery whether the route retries a failed message twice with
the default error handler (synchronous
+ * redelivery), after a delay longer than the visibility
timeout
+ */
+ SqsWorld(Dst dst, boolean redelivery) {
this.dst = dst;
+ this.redelivery = redelivery;
this.ledger = new DstLedger(dst);
this.sqs = new FakeSqs(dst, ledger, this::received,
extensionFaulted::addAll);
// 1 s is left out by default: the extender is always late with 1 s
(CAMEL-25737, open);
@@ -71,8 +90,11 @@ final class SqsWorld {
this.maxMessagesPerPoll = dst.pick(List.of(1, 2, 5, 10));
this.concurrentConsumers = dst.pick(List.of(1, 2, 3));
this.delay = dst.pick(List.of(100L, 500L, 1000L));
+ // only drawn with redelivery, so the seeds of the world without it
keep their configuration
+ this.redeliveryDelay = redelivery ? dst.between(visibilityTimeout *
1000L, 3L * visibilityTimeout * 1000) : 0;
dst.trace("world: visibilityTimeout=" + visibilityTimeout + "
maxMessagesPerPoll=" + maxMessagesPerPoll
- + " concurrentConsumers=" + concurrentConsumers + " delay="
+ delay);
+ + " concurrentConsumers=" + concurrentConsumers + " delay="
+ delay
+ + (redelivery ? " redeliveryDelay=" + redeliveryDelay : ""));
}
void run() throws Exception {
@@ -81,6 +103,11 @@ final class SqsWorld {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() {
+ if (redelivery) {
+ // synchronous redelivery: the error handler sleeps on the
consumer's thread between the attempts
+
errorHandler(defaultErrorHandler().maximumRedeliveries(MAX_REDELIVERIES)
+ .redeliveryDelay(redeliveryDelay));
+ }
from("aws2-sqs://dst?amazonSQSClient=#sqs&queueUrl=RAW(" +
FakeSqs.QUEUE_URL + ")"
+ "&extendMessageVisibility=true&visibilityTimeout=" +
visibilityTimeout
+ "&maxMessagesPerPoll=" + maxMessagesPerPoll +
"&concurrentConsumers=" + concurrentConsumers
@@ -88,6 +115,10 @@ final class SqsWorld {
.routeId(ROUTE)
.process(exchange -> {
String id =
exchange.getMessage().getHeader(Sqs2Constants.MESSAGE_ID, String.class);
+ int attempt =
exchange.getMessage().getHeader(Exchange.REDELIVERY_COUNTER, 0, Integer.class);
+ if (attempt > 0) {
+ redelivered(exchange.getExchangeId(), id,
attempt);
+ }
ledger.record("routed", "q", id);
// usually quick; sometimes longer than the
visibility timeout, which the extender covers
long work = faults && dst.chance(0.15)
@@ -95,11 +126,17 @@ final class SqsWorld {
: dst.between(1, 200);
// a blocking downstream call: the other consumers
and the extender run meanwhile
dst.block(work, "process " + id);
- release(id);
if (faults && dst.chance(0.05)) {
+ if (redelivery && attempt < MAX_REDELIVERIES) {
+ // the error handler retries it: the
consumer still holds the message
+ failedAt.put(exchange.getExchangeId(),
dst.now());
+ } else {
+ release(id);
+ }
ledger.record("failed", "q", id);
throw Dst.injected(new
IllegalStateException("dst: processing failed for " + id));
}
+ release(id);
ledger.record("completed", "q", id);
});
}
@@ -133,6 +170,10 @@ final class SqsWorld {
Invariants.everyOfferedRoutedAtLeastOnce(ledger, "sent",
"completed"),
Invariants.of("the queue is drained once the faults
cleared",
() -> sqs.allDeleted() ? null : "not deleted: " +
sqs.notDeleted()));
+ if (redelivery) {
+ dst.check(Invariants.of("a redelivery waits at least the
redelivery delay",
+ () -> redeliveredTooEarly.isEmpty() ? null :
redeliveredTooEarly.toString()));
+ }
} finally {
dst.dropPending();
context.stop();
@@ -171,6 +212,19 @@ final class SqsWorld {
held.put(m.id, holders + 1);
}
+ /**
+ * Camel's error handler tries a failed exchange again: the redelivery
delay must have passed in virtual time. By
+ * exchange, as a message whose visibility change was lost can be in
flight twice.
+ */
+ private void redelivered(String exchangeId, String id, int attempt) {
+ Long failed = failedAt.remove(exchangeId);
+ long waited = failed == null ? -1 : dst.now() - failed;
+ if (waited < redeliveryDelay) {
+ redeliveredTooEarly.add(id + " (attempt " + attempt + " after " +
waited + " ms, redeliveryDelay="
+ + redeliveryDelay + ")");
+ }
+ }
+
/** The route has finished with one delivery of the message. */
private void release(String id) {
held.computeIfPresent(id, (k, v) -> v > 1 ? v - 1 : null);
diff --git a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
index 0d13aa8d264a..16bf1c96d140 100644
--- a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
+++ b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
@@ -16,13 +16,16 @@ with the same step trace.
* Every thread pool created through the `ExecutorServiceManager` is backed by
one event loop on the test thread
(`DstThreadPoolFactory`). Tasks run one at a time; when several can run, the
seed picks one.
* Time is virtual: it jumps to the next due task. The time reads, sleeps and
blocking waits of the code under test are
- rewritten at test time (`@DstInstrument`), so "wait 30 seconds" costs
nothing.
+ rewritten at test time (`@DstInstrument`), and the Camel core code that
waits, times tasks or throttles reads the
+ virtual clock through its internal time source, so "wait 30 seconds" costs
nothing.
* A fake of the external system (a queue, a checkpoint store, a web API)
injects faults chosen by the same seed.
* Invariants are checked at the end of each run.
* `@DstTest(seeds = 1000)` runs a test method once per seed. A failing seed is
printed with its trace and is replayed
exactly with `-Ddst.seed=4711` or `@DstSeed(4711)`.
-This is a test tool. Nothing of it runs in production, and it changes nothing
in Camel.
+This is a test tool. Nothing of it runs in production. Camel core has one
internal hook for it, `TimeSource` in
+camel-util: with its default provider it reads `System` and sleeps with
`Thread.sleep`, and only an open run replaces
+the provider.
== A test
@@ -54,6 +57,8 @@ These worlds are in the Camel code base and are the examples
to copy:
|Component |Test |Regression test for
|camel-aws2-sqs |`SqsVisibilityDstTest` (the visibility timeout extender)
|CAMEL-25700
+|camel-aws2-sqs |`SqsRedeliveryDstTest` (the extender while Camel's error
handler waits to redeliver, in virtual
+time) |-
|camel-azure-eventhubs |`EventHubsCheckpointDstTest` (checkpointing by batch
size and timeout) |CAMEL-25683
|camel-pulsar |`PulsarConsumerDstTest` (subscribe failures, broker and route
restarts while the route starts and
consumes) |CAMEL-25869
@@ -78,13 +83,25 @@ virtual time. A wait that no task of the run can end fails
the seed at once and
|Timed waits on tasks |`Future.get(timeout)` on a task of the run and
`awaitTermination` time out in virtual time.
|Parallel streams |`parallelStream()` of a `java.util` collection is rewritten
to `stream()`: the elements are handled
in order by the calling task, not on the common fork-join pool.
+|Camel core |The code of Camel core that waits, times tasks or throttles reads
the clock and sleeps through
+`org.apache.camel.util.TimeSource`, an internal class of camel-util. A run
installs its virtual clock there (`Dst`
+does it when the run is created) and puts the previous provider back when the
run is closed, so no rewriting is needed:
+the synchronous redelivery (`RedeliveryPolicy`, `RedeliveryErrorHandler`),
`ForegroundTask`, `BackgroundTask.schedule`,
+the budgets of both tasks, the wait loop of `DefaultShutdownStrategy`, the
synchronous delayer, `MulticastProcessor`,
+`OptimisticLockRetryPolicy`, `DefaultStartupConditionStrategy`, `StopWatch`,
`MonotonicClock`, `BackOffTimerTask`,
+`DefaultTimeoutMap`, the clocks of the throttlers and the counters of
`ThrottlingExceptionRoutePolicy`. A sleep there is
+a virtual wait on the loop thread, traced with its call site. Three waits of
these classes are not virtual (see below):
+the permit queue of `TotalRequestsThrottler`, the `Semaphore` of
`ConcurrentRequestsThrottler` and the half-open timer
+of `ThrottlingExceptionRoutePolicy`. The sleep of the route `delayer` option
is not virtual either (see below).
|===
-`@DstInstrument(alsoRewrite = ...)` applies the same rewriting to classes
outside the code under test, for example
-`org.apache.camel.util.StopWatch`, `org.apache.camel.support.MonotonicClock`,
`org.apache.camel.support.task`
-(`ForegroundTask`), `org.apache.camel.processor.errorhandler` (synchronous
redelivery) or
-`org.apache.camel.impl.engine.DefaultShutdownStrategy` (its wait for in-flight
exchanges). This is a stop-gap: Camel
-core reads `System` directly, and a time source in core that the harness can
replace needs a dev@ discussion first.
+Event timestamps, the clock of an exchange and the direct clock reads of the
debugger, the tracer, management and the
+developer consoles stay on `System`. The durations they measure with
`StopWatch` or `MonotonicClock` follow `TimeSource`
+(with the default provider this is the same `System.nanoTime()`), and so does
the created time of `MonotonicClock`: the
+message history times (`MessageHistory.getTime()`) are virtual in a DST run.
`@DstInstrument(alsoRewrite = ...)` applies the
+rewriting of the code under test to classes outside it, for example
`org.apache.camel.support.ResetableClock` (the
+clock of an exchange) or a library the component calls. It is not needed for
the core classes listed above, except for
+the permit queue of `TotalRequestsThrottler` (see below).
=== Escapes and strict mode
@@ -93,15 +110,34 @@ Before the first seed, the harness scans the packages
under test and writes ever
cannot run the loop for (`Object.wait`, `LockSupport.park`, `Mono.block`),
wall clock reads it does not rewrite.
Strict mode is the default: when a seed reaches one of these at runtime, or a
thread outside the harness submits a task
-to the loop, reads the virtual clock or sleeps, the seed fails with the escape
as its cause. A clock read from such a
+to the loop, reads the virtual clock or sleeps, the seed fails with the escape
as its cause. This includes Camel core:
+a `StopWatch` started on such a thread reads the virtual clock through
`TimeSource`. A clock read from such a
thread (a Reactor or Netty thread, for example) still gets the virtual time
and its sleep still takes real time, so
that thread mixes the two; this is what strict mode is there to make visible.
`@DstInstrument(allowEscapes = "Foo.bar")` allows an escape
that cannot change the result (say why in a comment); `@DstInstrument(strict =
false)` or `-Ddst.strict=false` only
records escapes in `target/dst/<Test>.summary.txt`.
-A wait in code that is not rewritten (a latch in Camel core, for example)
blocks the loop thread for real. A watchdog
-interrupts the loop thread after 10 s of real time without progress
(`-Ddst.watchdogMillis`) and the seed fails with the
-stack where it was blocked.
+A wait in code that is not rewritten blocks the loop thread for real. A
watchdog interrupts the loop thread after 10 s
+of real time without progress (`-Ddst.watchdogMillis`) and the seed fails with
the stack where it was blocked. In Camel
+core:
+
+* `BackgroundTask.run` waits on a latch until its scheduled attempts are done,
so it blocks the loop thread.
+ `BackgroundTask.schedule` does not wait, and its attempts run in virtual
time.
+* `TotalRequestsThrottler` waits for a permit on its `DelayQueue`. The delay
of a permit follows the virtual clock,
+ which does not move while the loop thread waits, so a seed that has to wait
for a permit always ends on the watchdog.
+ `asyncDelayed()` does not avoid it: a delayed exchange that finds its permit
taken by another one waits on the same
+ queue. Add `org.apache.camel.processor.TotalRequestsThrottler` to
`@DstInstrument(alsoRewrite)`: the permit wait is
+ then a virtual wait.
+* `ConcurrentRequestsThrottler` blocks on its `Semaphore` once
`maxConcurrentRequests` exchanges are in flight. No task
+ of the run releases a permit while the loop thread waits, so the seed ends
on the watchdog. `alsoRewrite` does not
+ help here (the call goes through a `Semaphore` subclass): keep
`maxConcurrentRequests` above the number of exchanges
+ the world has in flight.
+* The route `delayer` option (`CamelInternalProcessor.DelayerAdvice`) calls
`Thread.sleep` directly, not through
+ `TimeSource`: the delay takes real time on the loop thread and the virtual
clock does not move. Use the `delay` EIP
+ in a world instead.
+* The half-open check of `ThrottlingExceptionRoutePolicy` runs on a
`java.util.Timer` thread, outside the harness. The
+ timer fires after `halfOpenAfter` of real time (30 s by default, longer than
the watchdog) and its clock read is an
+ escape, so a world does not see the circuit half-open in virtual time.
=== The one-stack model
@@ -404,12 +440,15 @@ seed of the in-JVM replay test of the Event Hubs world,
default 1).
* Code on its own threads or on a third-party scheduler is outside the
harness; strict mode makes reaching it fail the
seed.
* One run per JVM at a time: the active run, and so the virtual clock, is a
static field, and every rewritten clock read
- in the JVM sees it. A world must therefore not go into a module that runs
its tests in parallel
- (`camel.surefire.parallel=true`, as camel-jms and camel-sql do): the other
tests in that surefire fork would read the
- virtual clock while a run is active. The harness enforces this: a second run
while one is open fails, and a DST test
- refuses to start when JUnit runs tests in parallel. A local world in such a
module (camel-sjms, for example) can
- still be run alone, with `-Dtest=<the world>
-Dcamel.surefire.parallel=false`, but it cannot be committed there:
- the module's own build would fail it.
+ in the JVM sees it, as does Camel core through `TimeSource` while the run is
open. A world must therefore not go into
+ a module that runs its tests in parallel (`camel.surefire.parallel=true`, as
camel-jms and camel-sql do): the other
+ tests in that surefire fork would read the virtual clock while a run is
active. For the same reason, a
+ `CamelContext` left running in the same fork by an earlier non-DST test
would read the virtual time too and could
+ fail a strict seed; stop it in that test. The harness enforces this: a second
+ run while one is open fails, and a DST test refuses to start when JUnit runs
tests in parallel. A local world in
+ such a module (camel-sjms, for example) can still be run alone, with
+ `-Dtest=<the world> -Dcamel.surefire.parallel=false`, but it cannot be
committed there: the module's own build would
+ fail it.
* Locks do not exclude: every task runs on the test thread, so a
`ReentrantLock` or `synchronized` block held by a
blocked task is entered again by the tasks nested above it. A critical
section is atomic only because the fake does
not block inside it, and races between two code paths that each hold a lock
cannot be explored.
@@ -427,7 +466,7 @@ seed of the in-JVM replay test of the Event Hubs world,
default 1).
strict mode
|`DstThreadPoolFactory`, `DstExecutor`, `DstTask` |Camel's pools and
schedulers backed by the loop
|`DstInstrumentation`, `DstClock`, `DstWaits` |the byte code rewrite of the
code under test: time reads, sleeps and
-blocking waits go to the loop
+blocking waits go to the loop; `DstClock` is also the `TimeSource` provider of
Camel core during a run
|`DstEscapes`, `DstWatchdog` |the escape scan, runtime escapes (strict mode),
and the watchdog for waits in code that is
not rewritten
|`DstLedger`, `Invariant`, `Invariants` |what happened to each message, and
reusable invariants
@@ -446,7 +485,7 @@ not rewritten
|`dst.seed` |comma separated seeds to replay; their full traces are written |-
|`dst.traceDir` |where the reports and traces go |`target/dst`
|`dst.strict` |strict mode |`@DstInstrument(strict)`, true
-|`dst.alsoRewrite` |comma separated classes or packages to rewrite too
(stop-gap) |-
+|`dst.alsoRewrite` |comma separated classes or packages outside the code under
test to rewrite too |-
|`dst.watchdogMillis` |real time without progress after which a waiting loop
thread fails the seed |10000
|===
diff --git
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/Dst.java
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/Dst.java
index 1844447890cc..7df58495f69e 100644
---
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/Dst.java
+++
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/Dst.java
@@ -30,6 +30,8 @@ import java.util.Random;
import java.util.concurrent.Callable;
import java.util.function.BooleanSupplier;
+import org.apache.camel.util.TimeSource;
+
/**
* One deterministic simulation run: a single event loop that runs every task
of every executor created through
* {@link DstThreadPoolFactory}, a virtual clock, a seeded random source for
the scheduling choices and the faults, and
@@ -58,6 +60,8 @@ public final class Dst {
private static final int MAX_TRACE_LINES = 200_000;
private static volatile Dst active;
+ /** The {@link TimeSource} provider before the active run installed the
virtual clock, put back when it closes. */
+ private static TimeSource.Provider previousTimeSource;
private final long seed;
private final Random random;
@@ -114,8 +118,11 @@ public final class Dst {
/**
* Makes the run the active one. The virtual clock is JVM-wide (every
rewritten clock read in the JVM sees the
- * active run), so only one run may be open at a time: a second run while
another one is not closed (tests that run
- * in parallel, or a run that was not closed) fails fast instead of
sharing the clock.
+ * active run, and so does Camel core through {@link TimeSource}), so only
one run may be open at a time: a second
+ * run while another one is not closed (tests that run in parallel, or a
run that was not closed) fails fast instead
+ * of sharing the clock.
+ * <p>
+ * The virtual clock is installed as the {@link TimeSource} provider of
Camel core until the run is closed.
*/
private static synchronized void activate(Dst dst) {
Dst previous = active;
@@ -123,6 +130,9 @@ public final class Dst {
throw new IllegalStateException(
"DST: another run (seed " + previous.seed() + ") is still
active; one run per JVM at a time");
}
+ if (previous == null) {
+ previousTimeSource = TimeSource.install(DstClock.TIME_SOURCE);
+ }
active = dst;
}
@@ -640,7 +650,10 @@ public final class Dst {
pending.clear();
}
- /** Drops every pending task and detaches the virtual clock. */
+ /**
+ * Drops every pending task and detaches the virtual clock: Camel core
gets back the {@link TimeSource} provider it
+ * had before the run (the system clock, unless a test installed another
one).
+ */
public void close() {
synchronized (this) {
closed = true;
@@ -649,6 +662,8 @@ public final class Dst {
synchronized (Dst.class) {
if (active == this) {
active = null;
+ TimeSource.install(previousTimeSource);
+ previousTimeSource = null;
}
}
}
diff --git
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstClock.java
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstClock.java
index 0018badfbffa..bdb0bb6481c6 100644
---
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstClock.java
+++
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstClock.java
@@ -29,10 +29,17 @@ import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.concurrent.TimeUnit;
+import org.apache.camel.util.TimeSource;
+
/**
* The target of the rewritten time reads of the code under test (see {@link
DstInstrumentation}): the virtual clock of
* the active run, else the real clock.
* <p>
+ * The same clock is the {@link TimeSource} provider of Camel core while a run
is open ({@code TIME_SOURCE}, installed
+ * by {@link Dst}): the core code that waits, times tasks or throttles (the
redelivery delay, the wait loop of the
+ * graceful shutdown, {@code ForegroundTask}, {@code StopWatch}, the clocks of
the throttlers) reads the virtual time
+ * and sleeps virtual sleeps without any rewriting.
+ * <p>
* Rewritten: {@code System.currentTimeMillis()}, {@code System.nanoTime()},
{@code new Date()}, {@code Instant.now()},
* {@code Clock.systemUTC()}, {@code Clock.systemDefaultZone()}, {@code
Clock.system(ZoneId)} and {@code now()} /
* {@code now(ZoneId)} of {@code LocalDateTime}, {@code ZonedDateTime}, {@code
OffsetDateTime}, {@code LocalDate},
@@ -50,6 +57,37 @@ public final class DstClock {
private static final long NANO_OFFSET = 1_000_000_000_000L;
+ /**
+ * The {@link TimeSource} provider of Camel core while a run is open: the
same as the rewritten calls, with the
+ * caller of {@link TimeSource#sleep(long)} as the call site of a sleep in
the trace.
+ */
+ static final TimeSource.Provider TIME_SOURCE = new TimeSource.Provider() {
+ @Override
+ public long currentTimeMillis() {
+ return DstClock.currentTimeMillis();
+ }
+
+ @Override
+ public long nanoTime() {
+ return DstClock.nanoTime();
+ }
+
+ @Override
+ public Instant instant() {
+ return instantNow();
+ }
+
+ @Override
+ public void sleep(long millis) throws InterruptedException {
+ DstClock.sleep(millis, caller());
+ }
+
+ @Override
+ public String toString() {
+ return "DstClock";
+ }
+ };
+
private DstClock() {
}
@@ -78,6 +116,21 @@ public final class DstClock {
return dst == null ? Instant.now() :
Instant.ofEpochMilli(dst.currentTimeMillis());
}
+ /** The call site of a sleep through {@link TimeSource}, as the rewriting
writes it: {@code Class.method:line}. */
+ private static String caller() {
+ return StackWalker.getInstance().walk(frames -> frames
+ .filter(f -> !isTimeSourceFrame(f.getClassName()))
+ .findFirst()
+ .map(f ->
f.getClassName().substring(f.getClassName().lastIndexOf('.') + 1) + "." +
f.getMethodName()
+ + ":" + f.getLineNumber())
+ .orElse("TimeSource.sleep"));
+ }
+
+ private static boolean isTimeSourceFrame(String className) {
+ return className.equals(TimeSource.class.getName()) ||
className.startsWith(TimeSource.class.getName() + "$")
+ || className.equals(DstClock.class.getName()) ||
className.startsWith(DstClock.class.getName() + "$");
+ }
+
//
------------------------------------------------------------------------------------------------------------
// sleeps: Thread.sleep(long), Thread.sleep(long, int),
Thread.sleep(Duration) and TimeUnit.sleep(long), with the
// call site added by the rewriting
diff --git
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstExtension.java
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstExtension.java
index d2a90dd11619..48786ea04647 100644
---
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstExtension.java
+++
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstExtension.java
@@ -113,7 +113,7 @@ public class DstExtension implements
TestTemplateInvocationContextProvider, Befo
DstInstrumentation.alsoRewrite(System.getProperty("dst.alsoRewrite",
"").split(","));
StringBuilder sb = new StringBuilder();
if (!DstInstrumentation.alsoRewritten().isEmpty()) {
- sb.append("DST time reads also rewritten outside the code under
test (stop-gap): ")
+ sb.append("DST time reads also rewritten outside the code under
test: ")
.append(DstInstrumentation.alsoRewritten()).append('\n');
}
for (String p : instrument.value()) {
@@ -151,6 +151,14 @@ public class DstExtension implements
TestTemplateInvocationContextProvider, Befo
DstEscapes.loaded().forEach(e -> sb.append("
").append(e).append('\n'));
System.out.print(sb);
write(traceDir().resolve(context.getRequiredTestClass().getSimpleName() +
".summary.txt"), sb.toString());
+ Dst left = Dst.active();
+ if (left != null) {
+ // a run that was not closed keeps the virtual clock installed in
TimeSource for the later tests of the JVM
+ left.close();
+ throw new IllegalStateException(
+ "DST: the run of seed " + left.seed() + " was still open
after "
+ +
context.getRequiredTestClass().getSimpleName() + "; it is closed now");
+ }
}
@Override
@@ -313,16 +321,21 @@ public class DstExtension implements
TestTemplateInvocationContextProvider, Befo
stats.runs++;
stats.nanos += System.nanoTime() - start;
if (dst != null) {
- if (replay) {
- if (written == null) {
- written = writeTrace(failed ? "FAILED" : "PASSED");
+ try {
+ if (replay) {
+ if (written == null) {
+ written = writeTrace(failed ? "FAILED" : "PASSED");
+ }
+ String line = String.format("seed %d %s trace %s (%d
lines) -> %s", seed,
+ failed ? "FAILED" : "PASSED", dst.traceHash(),
dst.traceLines().size(), written);
+ stats.replayHashes.add(line);
+ System.out.println("DST replay " + line);
}
- String line = String.format("seed %d %s trace %s (%d
lines) -> %s", seed, failed ? "FAILED" : "PASSED",
- dst.traceHash(), dst.traceLines().size(), written);
- stats.replayHashes.add(line);
- System.out.println("DST replay " + line);
+ } finally {
+ // also when the trace cannot be written: an open run
keeps the virtual clock installed in
+ // TimeSource for the rest of the JVM
+ dst.close();
}
- dst.close();
}
}
diff --git
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstInstrument.java
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstInstrument.java
index 4b90a65cab53..9f9c9c558456 100644
---
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstInstrument.java
+++
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstInstrument.java
@@ -37,14 +37,19 @@ public @interface DstInstrument {
String[] value();
/**
- * Classes or packages outside the code under test whose time reads are
rewritten too, without an escape report, for
- * example {@code org.apache.camel.support.MonotonicClock} or {@code
org.apache.camel.util.StopWatch} of Camel core.
- * An entry whose last segment starts with an upper case letter is a class
(with its nested classes), else a
- * package.
+ * Classes or packages outside the code under test whose time reads,
sleeps and blocking waits are rewritten too,
+ * without an escape report, for example {@code
org.apache.camel.support.ResetableClock} of Camel core or a library
+ * the component calls. An entry whose last segment starts with an upper
case letter is a class (with its nested
+ * classes), else a package.
* <p>
- * This is a stop-gap: Camel core reads {@code System} directly, and the
rewriting of core classes stays for the
- * rest of the test JVM (outside a run the rewritten calls read the real
clock). The system property
- * {@code dst.alsoRewrite} (comma separated) adds entries without changing
the test.
+ * Not needed for the Camel core code that waits, times tasks or throttles
({@code StopWatch},
+ * {@code MonotonicClock}, the synchronous redelivery, {@code
ForegroundTask}, the wait loop of the graceful
+ * shutdown, the clocks of the throttlers): it reads the clock and sleeps
through
+ * {@code org.apache.camel.util.TimeSource}, whose provider is the virtual
clock while a run is open. The permit
+ * wait of {@code org.apache.camel.processor.TotalRequestsThrottler} is a
{@code DelayQueue.take()}, which is not a
+ * clock read: list that class when a world uses it. The rewriting stays
for the rest of the test JVM (outside a run
+ * the rewritten calls read the real clock). The system property {@code
dst.alsoRewrite} (comma separated) adds
+ * entries without changing the test.
*/
String[] alsoRewrite() default {};
diff --git
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstInstrumentation.java
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstInstrumentation.java
index df260e69d3ef..e0d172978882 100644
---
a/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstInstrumentation.java
+++
b/components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstInstrumentation.java
@@ -69,8 +69,10 @@ import net.bytebuddy.jar.asm.Opcodes;
* seed.</li>
* </ul>
* {@link #alsoRewrite(String...)} applies the same rewriting, without the
report, to classes outside the code under
- * test, for example {@code org.apache.camel.support.MonotonicClock} and
{@code org.apache.camel.util.StopWatch} of
- * Camel core. That is a stop-gap until Camel core has a time source the
harness can replace.
+ * test, for example {@code org.apache.camel.support.ResetableClock} of Camel
core. The core code that waits, times
+ * tasks or throttles reads the clock and sleeps through {@code
org.apache.camel.util.TimeSource}, whose provider is the
+ * virtual clock while a run is open (see {@link Dst}); only its blocking
waits, such as the permit queue of
+ * {@code org.apache.camel.processor.TotalRequestsThrottler}, still need the
rewriting.
* <p>
* Uses a runtime-attached agent (as Mockito's inline mock maker does) and the
ASM copy shipped in Byte Buddy.
*/
@@ -139,11 +141,11 @@ public final class DstInstrumentation {
}
/**
- * Rewrites the time reads of these classes or packages too, without
reporting their escape points: a stop-gap for
- * code outside the code under test that reads the clock, such as {@code
org.apache.camel.support.MonotonicClock} or
- * {@code org.apache.camel.util.StopWatch}. An entry whose last segment
starts with an upper case letter is a class
- * (with its nested classes), else a package (with its sub packages). The
rewriting stays for the rest of the JVM;
- * outside a run the rewritten calls read the real clock.
+ * Rewrites the time reads of these classes or packages too, without
reporting their escape points: for code outside
+ * the code under test that reads the clock and does not use {@code
org.apache.camel.util.TimeSource}, such as
+ * {@code org.apache.camel.support.ResetableClock}. An entry whose last
segment starts with an upper case letter is
+ * a class (with its nested classes), else a package (with its sub
packages). The rewriting stays for the rest of
+ * the JVM; outside a run the rewritten calls read the real clock.
*/
public static synchronized void alsoRewrite(String... classesOrPackages) {
List<String> added = new ArrayList<>();
diff --git
a/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstAlsoRewriteTest.java
b/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstAlsoRewriteTest.java
index c76b8d27a81f..0ab9ac52a789 100644
---
a/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstAlsoRewriteTest.java
+++
b/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstAlsoRewriteTest.java
@@ -16,49 +16,45 @@
*/
package org.apache.camel.test.dst;
-import org.apache.camel.support.MonotonicClock;
-import org.apache.camel.util.StopWatch;
+import org.apache.camel.support.ResetableClock;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
- * The stop-gap for time read outside the code under test: core classes listed
in {@link DstInstrument#alsoRewrite()}
- * read virtual time too.
+ * Time read outside the code under test and outside {@code TimeSource}: core
classes listed in
+ * {@link DstInstrument#alsoRewrite()} read virtual time too. {@code
ResetableClock} (the clock of an exchange) reads
+ * {@code System} directly, as event timestamps do.
*/
-@DstInstrument(value = "org.apache.camel.test.dst.sample",
- alsoRewrite = { "org.apache.camel.util.StopWatch",
"org.apache.camel.support.MonotonicClock" })
+@DstInstrument(value = "org.apache.camel.test.dst.sample", alsoRewrite =
"org.apache.camel.support.ResetableClock")
class DstAlsoRewriteTest {
@Test
void listedCoreClocksReadVirtualTime() {
-
assertTrue(DstInstrumentation.alsoRewritten().contains("org.apache.camel.util.StopWatch"),
+
assertTrue(DstInstrumentation.alsoRewritten().contains("org.apache.camel.support.ResetableClock"),
DstInstrumentation.alsoRewritten().toString());
Dst dst = new Dst(1);
try {
- StopWatch[] watch = new StopWatch[1];
- MonotonicClock[] clock = new MonotonicClock[1];
+ ResetableClock[] clock = new ResetableClock[1];
long[] taken = new long[2];
DstExecutor ex = dst.executor("ex", 1);
- ex.schedule("start", 1_000, () -> {
- watch[0] = new StopWatch();
- clock[0] = new MonotonicClock();
- });
+ ex.schedule("start", 1_000, () -> clock[0] = new ResetableClock());
ex.schedule("read", 3_601_000, () -> {
- taken[0] = watch[0].taken();
- taken[1] = clock[0].elapsed();
+ taken[0] = clock[0].elapsed();
+ clock[0].reset();
});
+ ex.schedule("read again", 3_601_500, () -> taken[1] =
clock[0].elapsed());
long start = System.nanoTime();
dst.runUntilQuiet();
assertEquals(3_600_000L, taken[0]);
- assertEquals(3_600_000L, taken[1]);
- assertEquals(Dst.EPOCH_MILLIS + 1_000, clock[0].getCreated());
+ assertEquals(500L, taken[1]);
+ assertEquals(Dst.EPOCH_MILLIS + 3_601_000, clock[0].getCreated());
assertTrue(System.nanoTime() - start < 5_000_000_000L, "an hour of
virtual time takes no real time");
} finally {
dst.close();
}
- // outside a run the rewritten classes read the real clock
- assertTrue(new StopWatch().taken() < 60_000);
+ // outside a run the rewritten class reads the real clock
+ assertTrue(new ResetableClock().elapsed() < 60_000);
}
}
diff --git
a/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstCoreWaitsTest.java
b/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstCoreWaitsTest.java
index bdea5f9bc63e..71eccb35cfa0 100644
---
a/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstCoreWaitsTest.java
+++
b/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstCoreWaitsTest.java
@@ -36,16 +36,13 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
- * The waits of Camel core that sleep on the calling thread, made virtual
through the stop-gap list of classes to
- * rewrite: the synchronous redelivery (RedeliveryPolicy.sleep and the sleep
loop of RedeliveryErrorHandler, which
- * measures with StopWatch), ForegroundTask, and the wait loop of
DefaultShutdownStrategy.
+ * The waits of Camel core that sleep on the calling thread run in virtual
time without any rewriting: core sleeps and
+ * reads the clock through {@code TimeSource}, and the run installs its
virtual clock there. Covered: the synchronous
+ * redelivery (RedeliveryPolicy.sleep and the sleep loop of
RedeliveryErrorHandler, which measures with StopWatch),
+ * ForegroundTask with an iteration and a time budget, the synchronous
delayer, and the wait loop of
+ * DefaultShutdownStrategy.
*/
-@DstInstrument(value = "org.apache.camel.test.dst.sample",
- alsoRewrite = {
- "org.apache.camel.processor.errorhandler",
- "org.apache.camel.support.task",
- "org.apache.camel.impl.engine.DefaultShutdownStrategy",
- "org.apache.camel.util.StopWatch" })
+@DstInstrument("org.apache.camel.test.dst.sample")
class DstCoreWaitsTest {
private static final long REAL_LIMIT = 5_000_000_000L;
@@ -168,6 +165,62 @@ class DstCoreWaitsTest {
}
}
+ @Test
+ void foregroundTaskGivesUpWhenItsTimeBudgetIsSpentInVirtualTime() {
+ Dst dst = new Dst(1);
+ try {
+ boolean[] completed = { true };
+ long[] finished = new long[1];
+ ForegroundTask[] task = new ForegroundTask[1];
+ dst.executor("waiter", 1).execute(() -> {
+ // the time budget (TimeBoundedBudget) ends the task, long
before the iteration budget would
+ task[0] = Tasks.foregroundTask().withName("never ready")
+
.withBudget(Budgets.iterationTimeBudget().withInterval(Duration.ofSeconds(1))
+
.withMaxDuration(Duration.ofSeconds(30)).withMaxIterations(1_000).build())
+ .build();
+ completed[0] = task[0].run(null, () -> false);
+ finished[0] = dst.now();
+ });
+ long start = System.nanoTime();
+ dst.runUntilQuiet();
+ assertFalse(completed[0]);
+ assertEquals(30_000, finished[0]);
+ assertEquals(Duration.ofSeconds(30), task[0].elapsed());
+ assertEquals(30, task[0].iteration());
+ assertTrue(System.nanoTime() - start < REAL_LIMIT);
+ } finally {
+ dst.close();
+ }
+ }
+
+ @Test
+ void synchronousDelayWaitsInVirtualTime() throws Exception {
+ Dst dst = new Dst(1);
+ DefaultCamelContext context = DstCamel.newContext(dst);
+ try {
+ long[] reached = { -1 };
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:in").routeId("r").delay(4_000).syncDelayed().process(e ->
reached[0] = dst.now());
+ }
+ });
+ context.start();
+ Exchange[] done = new Exchange[1];
+ dst.executor("client", 1).schedule("send", 1_000, () ->
send(context, "r", done));
+ long start = System.nanoTime();
+ dst.runUntilQuiet();
+ assertEquals(5_000, reached[0]);
+ assertNull(done[0].getException());
+ assertTraced(dst, "sleep at DelayProcessorSupport.sleep");
+ assertTrue(System.nanoTime() - start < REAL_LIMIT);
+ } finally {
+ dst.dropPending();
+ context.stop();
+ dst.close();
+ }
+ }
+
private static void assertTraced(Dst dst, String text) {
assertTrue(dst.traceLines().stream().anyMatch(l -> l.contains(text)),
"no trace line contains: " + text);
}
diff --git
a/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstTimeSourceTest.java
b/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstTimeSourceTest.java
new file mode 100644
index 000000000000..9f39d0e4ce81
--- /dev/null
+++
b/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstTimeSourceTest.java
@@ -0,0 +1,179 @@
+/*
+ * 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.test.dst;
+
+import java.time.Instant;
+import java.util.List;
+
+import org.apache.camel.support.MonotonicClock;
+import org.apache.camel.util.StopWatch;
+import org.apache.camel.util.TimeSource;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A run installs its virtual clock as the {@code TimeSource} provider of
Camel core and puts the previous provider back
+ * when it closes: the core clocks read virtual time without any rewriting.
+ */
+@DstInstrument("org.apache.camel.test.dst.sample")
+class DstTimeSourceTest {
+
+ @Test
+ void aRunInstallsTheVirtualClockUntilItIsClosed() {
+ assertTrue(TimeSource.isDefault());
+ Dst dst = new Dst(1);
+ try {
+ assertFalse(TimeSource.isDefault());
+ assertEquals(dst.currentTimeMillis(),
TimeSource.currentTimeMillis());
+ assertEquals(Instant.ofEpochMilli(dst.currentTimeMillis()),
TimeSource.instant());
+ long nanos = TimeSource.nanoTime();
+ dst.runFor(1_500);
+ assertEquals(1_500_000_000L, TimeSource.nanoTime() - nanos);
+ } finally {
+ dst.close();
+ }
+ assertTrue(TimeSource.isDefault(), "the system clock is back after the
run");
+ long real = System.currentTimeMillis();
+ assertTrue(Math.abs(TimeSource.currentTimeMillis() - real) < 60_000);
+ }
+
+ @Test
+ void coreClocksReadVirtualTimeWithoutRewriting() {
+
assertFalse(DstInstrumentation.alsoRewritten().contains("org.apache.camel.util.StopWatch"));
+ Dst dst = new Dst(1);
+ try {
+ StopWatch[] watch = new StopWatch[1];
+ MonotonicClock[] clock = new MonotonicClock[1];
+ long[] taken = new long[2];
+ DstExecutor ex = dst.executor("ex", 1);
+ ex.schedule("start", 1_000, () -> {
+ watch[0] = new StopWatch();
+ clock[0] = new MonotonicClock();
+ });
+ ex.schedule("read", 3_601_000, () -> {
+ taken[0] = watch[0].taken();
+ taken[1] = clock[0].elapsed();
+ });
+ long start = System.nanoTime();
+ dst.runUntilQuiet();
+ assertEquals(3_600_000L, taken[0]);
+ assertEquals(3_600_000L, taken[1]);
+ assertEquals(Dst.EPOCH_MILLIS + 1_000, clock[0].getCreated());
+ assertTrue(System.nanoTime() - start < 5_000_000_000L, "an hour of
virtual time takes no real time");
+ } finally {
+ dst.close();
+ }
+ // outside a run the core clocks read the real clock
+ assertTrue(new StopWatch().taken() < 60_000);
+ }
+
+ @Test
+ void aSleepThroughTheTimeSourceIsAVirtualWaitWithItsCallSite() {
+ Dst dst = new Dst(1);
+ try {
+ long[] woke = new long[1];
+ dst.executor("sleeper", 1).execute(() -> {
+ try {
+ TimeSource.sleep(7_000);
+ } catch (InterruptedException e) {
+ throw new IllegalStateException(e);
+ }
+ woke[0] = dst.now();
+ });
+ dst.executor("other", 1).schedule("meanwhile", 3_000, () ->
dst.trace("other task at " + dst.now()));
+ long start = System.nanoTime();
+ dst.runUntilQuiet();
+ assertEquals(7_000, woke[0]);
+ assertTrue(dst.traceLines().stream()
+ .anyMatch(l -> l.contains("block 7000ms sleep at
DstTimeSourceTest.lambda$")),
+ String.join("\n", dst.traceLines()));
+ assertTrue(dst.traceLines().stream().anyMatch(l ->
l.contains("other task at 3000")));
+ assertTrue(System.nanoTime() - start < 5_000_000_000L);
+ } finally {
+ dst.close();
+ }
+ }
+
+ @Test
+ void aSleepThroughTheTimeSourceKeepsTheSemanticsOfThreadSleep() {
+ Dst dst = new Dst(1);
+ try {
+ assertThrows(IllegalArgumentException.class, () ->
TimeSource.sleep(-1));
+ Thread.currentThread().interrupt();
+ assertThrows(InterruptedException.class, () ->
TimeSource.sleep(1_000));
+ assertFalse(Thread.currentThread().isInterrupted(), "the
interrupted status is cleared");
+ assertEquals(0, dst.now(), "no virtual time passed");
+ } finally {
+ Thread.interrupted();
+ dst.close();
+ }
+ }
+
+ @Test
+ void aCoreClockReadOffTheLoopIsAnEscape() throws Exception {
+ Dst dst = new Dst(1);
+ try {
+ dst.strict(true, List.of());
+ Thread outside = new Thread(StopWatch::new, "outside-watch");
+ outside.start();
+ outside.join();
+ assertTrue(dst.escape().contains("virtual clock read from a thread
outside the harness: outside-watch"),
+ dst.escape());
+ assertThrows(DstViolation.class, dst::runUntilQuiet);
+ } finally {
+ dst.close();
+ }
+ }
+
+ @Test
+ void theProviderInstalledBeforeTheRunIsPutBack() {
+ TimeSource.Provider fixed = new TimeSource.Provider() {
+ @Override
+ public long currentTimeMillis() {
+ return 42;
+ }
+
+ @Override
+ public long nanoTime() {
+ return 42;
+ }
+
+ @Override
+ public void sleep(long millis) {
+ // no sleep
+ }
+ };
+ TimeSource.Provider previous = TimeSource.install(fixed);
+ try {
+ Dst dst = new Dst(1);
+ try {
+ assertEquals(dst.currentTimeMillis(),
TimeSource.currentTimeMillis());
+ } finally {
+ dst.close();
+ }
+ assertEquals(42, TimeSource.currentTimeMillis());
+ } finally {
+ assertSame(fixed, TimeSource.install(previous));
+ }
+ assertTrue(TimeSource.isDefault());
+ }
+}
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
index 0c7b058cbfc1..dc45a9770d47 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
@@ -56,6 +56,7 @@ import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StopWatch;
+import org.apache.camel.util.TimeSource;
import org.apache.camel.util.TimeUtils;
import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
@@ -704,7 +705,7 @@ public class DefaultShutdownStrategy extends ServiceSupport
implements ShutdownS
// log verbose if DEBUG logging is enabled
logInflightExchanges(context, routes,
logInflightExchangesOnTimeout);
- Thread.sleep(loopDelaySeconds * 1000);
+ TimeSource.sleep(loopDelaySeconds * 1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
if (abortAfterTimeout) {
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/ConcurrentRequestsThrottler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/ConcurrentRequestsThrottler.java
index 2a5f9a0526c5..82e7033650ba 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/ConcurrentRequestsThrottler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/ConcurrentRequestsThrottler.java
@@ -36,6 +36,7 @@ import org.apache.camel.RouteStoppingException;
import org.apache.camel.RuntimeExchangeException;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -133,7 +134,7 @@ public class ConcurrentRequestsThrottler extends
AbstractThrottler {
// permit acquired
if (state == State.ASYNC) {
if (LOG.isTraceEnabled()) {
- long queuedTime = Duration.ofNanos(System.nanoTime() -
queuedStart).toMillis();
+ long queuedTime = Duration.ofNanos(TimeSource.nanoTime() -
queuedStart).toMillis();
LOG.trace("Queued for {}ms, No throttling applied
(throttle cleared while queued), for exchangeId: {}",
queuedTime, exchange.getExchangeId());
}
@@ -154,15 +155,15 @@ public class ConcurrentRequestsThrottler extends
AbstractThrottler {
long start = 0;
long elapsed = 0;
if (LOG.isTraceEnabled()) {
- start = System.nanoTime();
+ start = TimeSource.nanoTime();
}
throttlingState.acquire(exchange);
if (LOG.isTraceEnabled()) {
- elapsed = System.nanoTime() - start;
+ elapsed = Duration.ofNanos(TimeSource.nanoTime() -
start).toMillis();
}
if (state == State.ASYNC) {
if (LOG.isTraceEnabled()) {
- long queuedTime = start - queuedStart;
+ long queuedTime = Duration.ofNanos(start -
queuedStart).toMillis();
if (LOG.isTraceEnabled()) {
LOG.trace("Queued for {}ms, Throttled for {}ms,
exchangeId: {}", queuedTime, elapsed,
exchange.getExchangeId());
@@ -184,7 +185,7 @@ public class ConcurrentRequestsThrottler extends
AbstractThrottler {
final Exchange exchange, final AsyncCallback callback,
ThrottlingState throttlingState) {
try {
if (LOG.isTraceEnabled()) {
- exchange.setProperty(PROPERTY_EXCHANGE_QUEUED_TIME,
System.nanoTime());
+ exchange.setProperty(PROPERTY_EXCHANGE_QUEUED_TIME,
TimeSource.nanoTime());
}
exchange.setProperty(PROPERTY_EXCHANGE_STATE, State.ASYNC);
asyncExecutor.submit(() -> process(exchange, callback));
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
index 7193c9ab1a72..41163d794b60 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/DelayProcessorSupport.java
@@ -27,6 +27,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.RouteStoppingException;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -251,7 +252,7 @@ public abstract class DelayProcessorSupport extends
BaseDelegateProcessorSupport
}
protected long currentSystemTime() {
- return System.currentTimeMillis();
+ return TimeSource.currentTimeMillis();
}
private void sleep(long delay) throws InterruptedException {
@@ -259,7 +260,7 @@ public abstract class DelayProcessorSupport extends
BaseDelegateProcessorSupport
return;
}
LOG.trace("Sleeping for: {} millis", delay);
- Thread.sleep(delay);
+ TimeSource.sleep(delay);
}
@Override
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
index 243129e08ac9..c96b75160450 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
@@ -76,6 +76,7 @@ import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.CastUtils;
import org.apache.camel.util.IOHelper;
import org.apache.camel.util.StopWatch;
+import org.apache.camel.util.TimeSource;
import org.apache.camel.util.concurrent.AsyncCompletionService;
import org.apache.camel.util.concurrent.Rejectable;
import org.slf4j.Logger;
@@ -880,7 +881,7 @@ public class MulticastProcessor extends BaseProcessorSupport
} else {
executor.execute(() -> {
try {
- Thread.sleep(delay);
+ TimeSource.sleep(delay);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
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 c46d7ee12464..3e5a3b0d9b12 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
@@ -28,6 +28,7 @@ import org.apache.camel.Traceable;
import org.apache.camel.spi.IdAware;
import org.apache.camel.spi.RouteIdAware;
import org.apache.camel.spi.StepIdAware;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -145,8 +146,8 @@ public class SamplingThrottler extends BaseProcessorSupport
implements Traceable
doSend = true;
}
} else {
- long now = Duration.ofNanos(System.nanoTime()).toMillis();
- // System.nanoTime() has an arbitrary origin, so the first
exchange cannot be compared with a time
+ long now = Duration.ofNanos(TimeSource.nanoTime()).toMillis();
+ // nanoTime() has an arbitrary origin, so the first exchange
cannot be compared with a time
if (!sampledOnce || now >= timeOfLastExchange +
periodInMillis) {
sampledOnce = true;
doSend = true;
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 06363fa377cc..f949915439ad 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
@@ -35,6 +35,7 @@ import org.apache.camel.Expression;
import org.apache.camel.RouteStoppingException;
import org.apache.camel.RuntimeExchangeException;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -129,7 +130,7 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
long start = 0;
long elapsed = 0;
if (LOG.isTraceEnabled()) {
- start = System.currentTimeMillis();
+ start = TimeSource.currentTimeMillis();
}
permit = throttlingState.take();
while (throttlingState.isRemoved()) {
@@ -138,7 +139,7 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
permit = throttlingState.take();
}
if (LOG.isTraceEnabled()) {
- elapsed = System.currentTimeMillis() - start;
+ elapsed = TimeSource.currentTimeMillis() - start;
}
throttlingState.enqueue(permit, exchange);
@@ -161,7 +162,7 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
if (state == State.ASYNC) {
if (LOG.isTraceEnabled()) {
- long queuedTime = System.currentTimeMillis() -
queuedStart;
+ long queuedTime = TimeSource.currentTimeMillis() -
queuedStart;
LOG.trace("Queued for {}ms, No throttling applied
(throttle cleared while queued), for exchangeId: {}",
queuedTime, exchange.getExchangeId());
}
@@ -192,7 +193,7 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
final Exchange exchange, final AsyncCallback callback,
ThrottlingState throttlingState) {
try {
if (LOG.isTraceEnabled()) {
- exchange.setProperty(PROPERTY_EXCHANGE_QUEUED_TIMESTAMP,
System.nanoTime());
+ exchange.setProperty(PROPERTY_EXCHANGE_QUEUED_TIMESTAMP,
TimeSource.currentTimeMillis());
}
exchange.setProperty(PROPERTY_EXCHANGE_STATE, State.ASYNC);
ThrottlePermit next = throttlingState.peek();
@@ -413,12 +414,12 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
}
public void setDelayMs(final long delayMs) {
- this.scheduledTime = System.currentTimeMillis() + delayMs;
+ this.scheduledTime = TimeSource.currentTimeMillis() + delayMs;
}
@Override
public long getDelay(final TimeUnit unit) {
- return unit.convert(scheduledTime - System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
+ return unit.convert(scheduledTime -
TimeSource.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/OptimisticLockRetryPolicy.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/OptimisticLockRetryPolicy.java
index 24a0032c9196..aee8b1fd6c9f 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/OptimisticLockRetryPolicy.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/OptimisticLockRetryPolicy.java
@@ -18,6 +18,8 @@ package org.apache.camel.processor.aggregate;
import java.util.concurrent.ThreadLocalRandom;
+import org.apache.camel.util.TimeSource;
+
/**
* Class to control how failed optimistic locks are tried. This policy
supports random and exponential back-off delays.
* <p/>
@@ -60,7 +62,7 @@ public class OptimisticLockRetryPolicy {
public void doDelay(final int retryCounter) throws InterruptedException {
long sleepFor = getDelay(retryCounter);
if (sleepFor > 0) {
- Thread.sleep(sleepFor);
+ TimeSource.sleep(sleepFor);
}
}
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
index f0281e9857b6..99e655477a45 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
@@ -59,6 +59,7 @@ import org.apache.camel.support.PluginHelper;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StopWatch;
+import org.apache.camel.util.TimeSource;
import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -2075,7 +2076,7 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
long max = Math.min(1000, delta);
if (max > 0) {
LOG.trace("Sleeping for: {} millis until waking up for
re-check", max);
- Thread.sleep(max);
+ TimeSource.sleep(max);
}
// are we preparing for shutdown then only do redelivery if
allowed
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryPolicy.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryPolicy.java
index ffb3605f4375..21ae719c3cfc 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryPolicy.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryPolicy.java
@@ -28,6 +28,7 @@ import org.apache.camel.Predicate;
import org.apache.camel.spi.Configurer;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StringHelper;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -205,7 +206,7 @@ public class RedeliveryPolicy implements Cloneable,
Serializable {
*/
public void sleep(long redeliveryDelay) throws InterruptedException {
LOG.debug("Sleeping for: {} millis until attempting redelivery",
redeliveryDelay);
- Thread.sleep(redeliveryDelay);
+ TimeSource.sleep(redeliveryDelay);
}
/**
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
b/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
index e64f980078df..3e3ac2fe785b 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
@@ -30,6 +30,7 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.TimeoutMap;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -268,7 +269,7 @@ public class DefaultTimeoutMap<K, V> extends ServiceSupport
implements TimeoutMa
}
protected long currentTime() {
- return System.currentTimeMillis();
+ return TimeSource.currentTimeMillis();
}
@Override
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/MonotonicClock.java
b/core/camel-support/src/main/java/org/apache/camel/support/MonotonicClock.java
index 01ab4f8beb89..1cde49f88fd1 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/MonotonicClock.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/MonotonicClock.java
@@ -19,6 +19,7 @@ package org.apache.camel.support;
import java.util.concurrent.TimeUnit;
import org.apache.camel.clock.Clock;
+import org.apache.camel.util.TimeSource;
/**
* A clock that increases monotonically (i.e.: does not go back in time) for
precise elapsed calculations.
@@ -30,13 +31,13 @@ public final class MonotonicClock implements Clock {
private final long createdNano;
public MonotonicClock() {
- this.created = System.currentTimeMillis();
- this.createdNano = System.nanoTime();
+ this.created = TimeSource.currentTimeMillis();
+ this.createdNano = TimeSource.nanoTime();
}
@Override
public long elapsed() {
- return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - createdNano);
+ return TimeUnit.NANOSECONDS.toMillis(TimeSource.nanoTime() -
createdNano);
}
@Override
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/startup/DefaultStartupConditionStrategy.java
b/core/camel-support/src/main/java/org/apache/camel/support/startup/DefaultStartupConditionStrategy.java
index de7c8aeeafab..6e5edd3eab1b 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/startup/DefaultStartupConditionStrategy.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/startup/DefaultStartupConditionStrategy.java
@@ -31,6 +31,7 @@ import org.apache.camel.spi.StartupStepRecorder;
import org.apache.camel.support.OrderedComparator;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.StopWatch;
+import org.apache.camel.util.TimeSource;
import org.apache.camel.util.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -195,7 +196,7 @@ public class DefaultStartupConditionStrategy extends
ServiceSupport implements S
// wait a bit before next loop
try {
- Thread.sleep(interval);
+ TimeSource.sleep(interval);
} catch (InterruptedException e) {
if (LOG.isDebugEnabled()) {
LOG.debug("Sleep interrupted, are we stopping? {}",
isCamelStopping());
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/task/AbstractTask.java
b/core/camel-support/src/main/java/org/apache/camel/support/task/AbstractTask.java
index 2ea8f365c8a9..4d14a2c017ce 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/task/AbstractTask.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/task/AbstractTask.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.support.task;
+import org.apache.camel.util.TimeSource;
+
abstract class AbstractTask implements Task {
static final long NEVER = -1L;
@@ -63,7 +65,7 @@ abstract class AbstractTask implements Task {
@Override
public long getCurrentElapsedTime() {
if (firstAttemptTime > 0) {
- return System.currentTimeMillis() - firstAttemptTime;
+ return TimeSource.currentTimeMillis() - firstAttemptTime;
}
return NEVER;
}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
b/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
index 41fade98bf95..834a9affe11c 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
@@ -31,6 +31,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.support.PluginHelper;
import org.apache.camel.support.task.budget.TimeBoundedBudget;
import org.apache.camel.support.task.budget.TimeBudget;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -128,7 +129,7 @@ public class BackgroundTask extends AbstractTask implements
BlockingTask {
return;
}
- lastAttemptTime = System.currentTimeMillis();
+ lastAttemptTime = TimeSource.currentTimeMillis();
if (firstAttemptTime < 0) {
firstAttemptTime = lastAttemptTime;
}
@@ -156,7 +157,7 @@ public class BackgroundTask extends AbstractTask implements
BlockingTask {
throw e;
}
// scheduleWithFixedDelay waits interval after this run finishes, so
compute from now
- nextAttemptTime = System.currentTimeMillis() + budget.interval();
+ nextAttemptTime = TimeSource.currentTimeMillis() + budget.interval();
}
/**
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
b/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
index fc423a3c2a65..1ae2983fb91f 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
@@ -26,6 +26,7 @@ import java.util.function.Supplier;
import org.apache.camel.CamelContext;
import org.apache.camel.support.PluginHelper;
import org.apache.camel.support.task.budget.IterationBudget;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -96,11 +97,11 @@ public class ForegroundTask extends AbstractTask implements
BlockingTask {
}
try {
if (budget.initialDelay() > 0) {
- Thread.sleep(budget.initialDelay());
+ TimeSource.sleep(budget.initialDelay());
}
while (budget.next()) {
- lastAttemptTime = System.currentTimeMillis();
+ lastAttemptTime = TimeSource.currentTimeMillis();
if (firstAttemptTime < 0) {
firstAttemptTime = lastAttemptTime;
}
@@ -114,7 +115,7 @@ public class ForegroundTask extends AbstractTask implements
BlockingTask {
}
if (budget.canContinue()) {
- Thread.sleep(budget.interval());
+ TimeSource.sleep(budget.interval());
} else {
status = Status.Exhausted;
}
@@ -169,11 +170,11 @@ public class ForegroundTask extends AbstractTask
implements BlockingTask {
}
try {
if (budget.initialDelay() > 0) {
- Thread.sleep(budget.initialDelay());
+ TimeSource.sleep(budget.initialDelay());
}
while (budget.next()) {
- lastAttemptTime = System.currentTimeMillis();
+ lastAttemptTime = TimeSource.currentTimeMillis();
if (firstAttemptTime < 0) {
firstAttemptTime = lastAttemptTime;
}
@@ -187,7 +188,7 @@ public class ForegroundTask extends AbstractTask implements
BlockingTask {
nextAttemptTime = lastAttemptTime + budget.interval();
if (budget.canContinue()) {
- Thread.sleep(budget.interval());
+ TimeSource.sleep(budget.interval());
} else {
status = Status.Exhausted;
}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudget.java
b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudget.java
index 7acc0f5f6ecf..6f4cde765d38 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudget.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudget.java
@@ -21,6 +21,7 @@ import java.time.Instant;
import org.apache.camel.support.task.budget.backoff.BackOffStrategy;
import org.apache.camel.support.task.budget.backoff.FixedBackOffStrategy;
+import org.apache.camel.util.TimeSource;
/**
* This task budget limits the execution by a given number of iterations or an
unlimited number if configured to do so.
@@ -42,14 +43,14 @@ public class IterationBoundedBudget implements
IterationBudget {
this.initialDelay = initialDelay;
this.maxIterations = maxIterations;
this.backOffStrategy = new FixedBackOffStrategy(interval);
- this.startTime = Instant.now();
+ this.startTime = TimeSource.instant();
}
IterationBoundedBudget(long initialDelay, int maxIterations,
BackOffStrategy backOffStrategy) {
this.initialDelay = initialDelay;
this.maxIterations = maxIterations;
this.backOffStrategy = backOffStrategy;
- this.startTime = Instant.now();
+ this.startTime = TimeSource.instant();
}
@Override
@@ -92,6 +93,6 @@ public class IterationBoundedBudget implements
IterationBudget {
@Override
public Duration elapsed() {
- return Duration.between(startTime, Instant.now());
+ return Duration.between(startTime, TimeSource.instant());
}
}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBoundedBudget.java
b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBoundedBudget.java
index 6606d041a545..40ce06482bfb 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBoundedBudget.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBoundedBudget.java
@@ -19,6 +19,8 @@ package org.apache.camel.support.task.budget;
import java.time.Duration;
import java.time.Instant;
+import org.apache.camel.util.TimeSource;
+
/**
* This task budget limits the execution by both a maximum amount of time for
the execution.
*/
@@ -36,7 +38,7 @@ public class TimeBoundedBudget implements TimeBudget {
this.initialDelay = initialDelay;
this.interval = interval;
this.maxDuration = maxDuration;
- this.startTime = Instant.now();
+ this.startTime = TimeSource.instant();
}
@Override
@@ -82,6 +84,6 @@ public class TimeBoundedBudget implements TimeBudget {
@Override
public Duration elapsed() {
- return Duration.between(startTime, Instant.now());
+ return Duration.between(startTime, TimeSource.instant());
}
}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
b/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
index 1f10d577692b..ab9e81f85e1f 100644
---
a/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
+++
b/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
@@ -36,6 +36,7 @@ import org.apache.camel.spi.Configurer;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.RoutePolicy;
import org.apache.camel.support.RoutePolicySupport;
+import org.apache.camel.util.TimeSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -203,7 +204,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
if (hasFailed(exchange)) {
// record the failure
failures.incrementAndGet();
- lastFailure = System.currentTimeMillis();
+ lastFailure = TimeSource.currentTimeMillis();
} else {
success.incrementAndGet();
}
@@ -269,7 +270,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
}
} else if (state.get() == STATE_OPEN) {
if (!keepOpenBool.get()) {
- long elapsedTimeSinceOpened = System.currentTimeMillis() -
openedAt;
+ long elapsedTimeSinceOpened = TimeSource.currentTimeMillis() -
openedAt;
if (halfOpenAfter <= elapsedTimeSinceOpened) {
LOG.debug("Checking an open circuit...");
if (halfOpenHandler != null) {
@@ -301,7 +302,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
logState();
// failures exceed the threshold
// AND the last of those failures occurred within window
- if (failures.get() >= failureThreshold && lastFailure >=
System.currentTimeMillis() - failureWindow) {
+ if (failures.get() >= failureThreshold && lastFailure >=
TimeSource.currentTimeMillis() - failureWindow) {
output = true;
}
@@ -316,7 +317,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
}
suspendOrStopConsumer(route.getConsumer());
state.set(STATE_OPEN);
- openedAt = System.currentTimeMillis();
+ openedAt = TimeSource.currentTimeMillis();
this.addHalfOpenTimer(route);
logState();
} catch (Exception e) {
@@ -336,7 +337,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
lock.lock();
suspendOrStopConsumer(route.getConsumer());
state.set(STATE_OPEN);
- openedAt = System.currentTimeMillis();
+ openedAt = TimeSource.currentTimeMillis();
this.addHalfOpenTimer(route);
logState();
} catch (Exception e) {
@@ -414,7 +415,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
String routeState = getStateAsString();
if (failures.get() > 0) {
return String.format("State %s, failures %d, last failure %d ms
ago", routeState, failures.get(),
- System.currentTimeMillis() - lastFailure);
+ TimeSource.currentTimeMillis() - lastFailure);
} else {
return String.format("State %s, failures %d", routeState,
failures.get());
}
diff --git a/core/camel-util/src/main/java/org/apache/camel/util/StopWatch.java
b/core/camel-util/src/main/java/org/apache/camel/util/StopWatch.java
index be49ceeba714..9e92ac81b0c2 100644
--- a/core/camel-util/src/main/java/org/apache/camel/util/StopWatch.java
+++ b/core/camel-util/src/main/java/org/apache/camel/util/StopWatch.java
@@ -31,7 +31,7 @@ public final class StopWatch {
* Starts the stop watch
*/
public StopWatch() {
- this.start = System.nanoTime();
+ this.start = TimeSource.nanoTime();
}
/**
@@ -41,7 +41,7 @@ public final class StopWatch {
*/
public StopWatch(boolean start) {
if (start) {
- this.start = System.nanoTime();
+ this.start = TimeSource.nanoTime();
}
}
@@ -49,7 +49,7 @@ public final class StopWatch {
* Starts or restarts the stop watch
*/
public void restart() {
- start = System.nanoTime();
+ start = TimeSource.nanoTime();
}
/**
@@ -66,7 +66,7 @@ public final class StopWatch {
*/
public long taken() {
if (start > 0) {
- long delta = System.nanoTime() - start;
+ long delta = TimeSource.nanoTime() - start;
return Duration.ofNanos(delta).toMillis();
}
return 0;
@@ -79,7 +79,7 @@ public final class StopWatch {
*/
public long takenAndRestart() {
long answer = taken();
- start = System.nanoTime();
+ start = TimeSource.nanoTime();
return answer;
}
diff --git
a/core/camel-util/src/main/java/org/apache/camel/util/TimeSource.java
b/core/camel-util/src/main/java/org/apache/camel/util/TimeSource.java
new file mode 100644
index 000000000000..de67666aaa8c
--- /dev/null
+++ b/core/camel-util/src/main/java/org/apache/camel/util/TimeSource.java
@@ -0,0 +1,172 @@
+/*
+ * 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.util;
+
+import java.time.Instant;
+
+/**
+ * The clock and the sleeps of the Camel core code that waits, times tasks or
throttles (for example the redelivery
+ * delay, the wait loop of the graceful shutdown, {@link StopWatch} and the
throttlers).
+ * <p>
+ * <b>Internal:</b> this is not an API for Camel users or components. The
default provider delegates to
+ * {@link System#currentTimeMillis()}, {@link System#nanoTime()}, {@link
Instant#now()} and {@link Thread#sleep(long)}.
+ * Only a test harness, such as the deterministic simulation tests of
camel-test-dst, installs another provider with
+ * {@link #install(Provider)}, for example a virtual clock, and puts the
previous one back when the test is done.
+ * <p>
+ * The provider is static rather than per {@code CamelContext}, because {@link
StopWatch} and the other timers have no
+ * context. Event timestamps and the direct clock reads of the debugger, the
tracer, management and the developer
+ * consoles stay on {@link System}; the durations they measure with {@link
StopWatch} or {@code MonotonicClock} follow
+ * this class (with the default provider this is the same {@link
System#nanoTime()}). The created time of
+ * {@code MonotonicClock} follows this class too, so the message history times
({@code MessageHistory.getTime()}) are
+ * virtual in a deterministic simulation test.
+ */
+public final class TimeSource {
+
+ /**
+ * The clock and the sleep behind {@link TimeSource}. Implemented only by
test harnesses.
+ */
+ public interface Provider {
+
+ /**
+ * The current time in milliseconds since the epoch, as {@link
System#currentTimeMillis()}.
+ */
+ long currentTimeMillis();
+
+ /**
+ * A monotonic time in nanoseconds with an arbitrary origin, as {@link
System#nanoTime()}: only the difference
+ * of two values means something.
+ */
+ long nanoTime();
+
+ /**
+ * The current instant, as {@link Instant#now()}. By default {@link
#currentTimeMillis()} as an instant.
+ */
+ default Instant instant() {
+ return Instant.ofEpochMilli(currentTimeMillis());
+ }
+
+ /**
+ * Sleeps for the given milliseconds, as {@link Thread#sleep(long)}.
+ *
+ * @throws IllegalArgumentException if the value of millis is negative
+ * @throws InterruptedException if the thread is interrupted
before or while it sleeps; the interrupted
+ * status of the thread is cleared
when this exception is thrown
+ */
+ void sleep(long millis) throws InterruptedException;
+ }
+
+ private static final Provider SYSTEM = new SystemProvider();
+
+ private static volatile Provider provider = SYSTEM;
+
+ private TimeSource() {
+ }
+
+ /**
+ * The current time in milliseconds since the epoch: {@link
System#currentTimeMillis()} unless a test harness
+ * installed another provider.
+ */
+ public static long currentTimeMillis() {
+ return provider.currentTimeMillis();
+ }
+
+ /**
+ * A monotonic time in nanoseconds with an arbitrary origin: {@link
System#nanoTime()} unless a test harness
+ * installed another provider. Only the difference of two values means
something.
+ */
+ public static long nanoTime() {
+ return provider.nanoTime();
+ }
+
+ /**
+ * The current instant: {@link Instant#now()} unless a test harness
installed another provider.
+ */
+ public static Instant instant() {
+ return provider.instant();
+ }
+
+ /**
+ * Sleeps for the given milliseconds: {@link Thread#sleep(long)} unless a
test harness installed another provider.
+ *
+ * @throws IllegalArgumentException if the value of millis is negative
+ * @throws InterruptedException if the thread is interrupted before or
while it sleeps; the interrupted status
+ * of the thread is cleared when this
exception is thrown
+ */
+ public static void sleep(long millis) throws InterruptedException {
+ provider.sleep(millis);
+ }
+
+ /**
+ * Installs a provider, for test harnesses only.
+ *
+ * @param provider the provider to use from now on
+ * @return the provider that was installed before, to be
installed again when the test is done
+ */
+ public static Provider install(Provider provider) {
+ ObjectHelper.notNull(provider, "provider");
+ synchronized (TimeSource.class) {
+ Provider previous = TimeSource.provider;
+ TimeSource.provider = provider;
+ return previous;
+ }
+ }
+
+ /**
+ * Installs the default provider again ({@link System} and {@link
Thread#sleep(long)}), for test harnesses only.
+ */
+ public static void reset() {
+ install(SYSTEM);
+ }
+
+ /**
+ * Whether the default provider ({@link System} and {@link
Thread#sleep(long)}) is installed.
+ */
+ public static boolean isDefault() {
+ return provider == SYSTEM;
+ }
+
+ /**
+ * The default provider: {@link System} and {@link Thread#sleep(long)}.
+ */
+ private static final class SystemProvider implements Provider {
+
+ @Override
+ public long currentTimeMillis() {
+ return System.currentTimeMillis();
+ }
+
+ @Override
+ public long nanoTime() {
+ return System.nanoTime();
+ }
+
+ @Override
+ public Instant instant() {
+ return Instant.now();
+ }
+
+ @Override
+ public void sleep(long millis) throws InterruptedException {
+ Thread.sleep(millis);
+ }
+
+ @Override
+ public String toString() {
+ return "SystemTimeSource";
+ }
+ }
+}
diff --git
a/core/camel-util/src/main/java/org/apache/camel/util/backoff/BackOffTimerTask.java
b/core/camel-util/src/main/java/org/apache/camel/util/backoff/BackOffTimerTask.java
index 27015c0c3505..6e0c3d71198a 100644
---
a/core/camel-util/src/main/java/org/apache/camel/util/backoff/BackOffTimerTask.java
+++
b/core/camel-util/src/main/java/org/apache/camel/util/backoff/BackOffTimerTask.java
@@ -26,6 +26,7 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
+import org.apache.camel.util.TimeSource;
import org.apache.camel.util.function.ThrowingFunction;
public final class BackOffTimerTask implements BackOffTimer.Task, Runnable {
@@ -171,7 +172,7 @@ public final class BackOffTimerTask implements
BackOffTimer.Task, Runnable {
public void run() {
if (status == Status.Active) {
try {
- lastAttemptTime = System.currentTimeMillis();
+ lastAttemptTime = TimeSource.currentTimeMillis();
if (firstAttemptTime < 0) {
firstAttemptTime = lastAttemptTime;
}
diff --git
a/core/camel-util/src/test/java/org/apache/camel/util/TimeSourceTest.java
b/core/camel-util/src/test/java/org/apache/camel/util/TimeSourceTest.java
new file mode 100644
index 000000000000..3ac078fbe3d3
--- /dev/null
+++ b/core/camel-util/src/test/java/org/apache/camel/util/TimeSourceTest.java
@@ -0,0 +1,162 @@
+/*
+ * 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.util;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Isolated;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The provider of {@link TimeSource} is JVM-wide: the tests that install one
run isolated from the other tests.
+ */
+@Isolated
+class TimeSourceTest {
+
+ /** A clock that only moves when asked to, and records the sleeps. */
+ private static final class ManualProvider implements TimeSource.Provider {
+ long millis = 1_000_000L;
+ final List<Long> sleeps = new ArrayList<>();
+
+ @Override
+ public long currentTimeMillis() {
+ return millis;
+ }
+
+ @Override
+ public long nanoTime() {
+ return millis * 1_000_000L;
+ }
+
+ @Override
+ public void sleep(long millis) {
+ sleeps.add(millis);
+ this.millis += millis;
+ }
+ }
+
+ @AfterEach
+ void restoreDefault() {
+ TimeSource.reset();
+ }
+
+ @Test
+ void defaultProviderReadsTheSystemClock() {
+ assertTrue(TimeSource.isDefault());
+ long before = System.currentTimeMillis();
+ long millis = TimeSource.currentTimeMillis();
+ long after = System.currentTimeMillis();
+ assertTrue(before <= millis && millis <= after, before + " <= " +
millis + " <= " + after);
+
+ long nanosBefore = System.nanoTime();
+ long nanos = TimeSource.nanoTime();
+ assertTrue(nanos - nanosBefore >= 0);
+ assertTrue(System.nanoTime() - nanos >= 0);
+
+ Instant instantBefore = Instant.now();
+ Instant instant = TimeSource.instant();
+ assertFalse(instant.isBefore(instantBefore));
+ assertFalse(instant.isAfter(Instant.now()));
+ }
+
+ @Test
+ void defaultSleepReturnsForZero() throws InterruptedException {
+ StopWatch watch = new StopWatch();
+ TimeSource.sleep(0);
+ assertTrue(watch.taken() < 1_000, "took " + watch.taken());
+ }
+
+ @Test
+ void defaultSleepRejectsANegativeValueAsThreadSleep() {
+ assertThrows(IllegalArgumentException.class, () ->
TimeSource.sleep(-1));
+ }
+
+ @Test
+ void defaultSleepOfAnInterruptedThreadThrowsAndClearsTheFlag() {
+ Thread.currentThread().interrupt();
+ try {
+ assertThrows(InterruptedException.class, () ->
TimeSource.sleep(10_000));
+ assertFalse(Thread.currentThread().isInterrupted(), "the
interrupted status is cleared, as Thread.sleep");
+ } finally {
+ // never leave the flag set for the next test
+ Thread.interrupted();
+ }
+ }
+
+ @Test
+ void installedProviderIsUsedUntilReset() throws InterruptedException {
+ ManualProvider manual = new ManualProvider();
+ TimeSource.Provider previous = TimeSource.install(manual);
+ assertNotSame(manual, previous);
+ assertFalse(TimeSource.isDefault());
+
+ assertEquals(1_000_000L, TimeSource.currentTimeMillis());
+ assertEquals(1_000_000L * 1_000_000L, TimeSource.nanoTime());
+ assertEquals(Instant.ofEpochMilli(1_000_000L), TimeSource.instant());
+
+ TimeSource.sleep(2_500);
+ assertEquals(List.of(2_500L), manual.sleeps);
+ assertEquals(1_002_500L, TimeSource.currentTimeMillis());
+
+ TimeSource.reset();
+ assertTrue(TimeSource.isDefault());
+ TimeSource.sleep(0);
+ assertEquals(List.of(2_500L), manual.sleeps, "the default provider is
back");
+ }
+
+ @Test
+ void installReturnsThePreviousProviderToPutBack() {
+ ManualProvider first = new ManualProvider();
+ ManualProvider second = new ManualProvider();
+ second.millis = 42;
+ TimeSource.Provider original = TimeSource.install(first);
+ assertSame(first, TimeSource.install(second));
+ assertEquals(42, TimeSource.currentTimeMillis());
+ assertSame(second, TimeSource.install(first));
+ assertEquals(1_000_000L, TimeSource.currentTimeMillis());
+ assertSame(first, TimeSource.install(original));
+ assertTrue(TimeSource.isDefault());
+ }
+
+ @Test
+ void installRejectsNull() {
+ assertThrows(IllegalArgumentException.class, () ->
TimeSource.install(null));
+ assertTrue(TimeSource.isDefault());
+ }
+
+ @Test
+ void stopWatchReadsTheInstalledProvider() {
+ ManualProvider manual = new ManualProvider();
+ TimeSource.install(manual);
+ StopWatch watch = new StopWatch();
+ manual.millis += 3_600_000L;
+ assertEquals(3_600_000L, watch.taken());
+ assertEquals(3_600_000L, watch.takenAndRestart());
+ manual.millis += 5;
+ assertEquals(5, watch.taken());
+ }
+}
