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 62c0f8e1b1dc CAMEL-26456: camel-chaos - a DST world that runs chaos
rules in virtual time (#29046)
62c0f8e1b1dc is described below
commit 62c0f8e1b1dc598996c32a1d88a70b2ca1af5d9e
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Nov 5 00:42:14 2026 +0530
CAMEL-26456: camel-chaos - a DST world that runs chaos rules in virtual
time (#29046)
A deterministic simulation (DST) world for the camel-chaos route lifecycle
rules: a fake queue on a real ScheduledPollConsumer runs the stall and
lifecycle (suspend, stop, remove) faults under the seeded virtual-time
scheduler and checks the invariants, including the redeliveries count.
Test-only; camel-test-dst is a test dependency.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/chaos.adoc | 7 +
components/camel-chaos/pom.xml | 25 ++
components/camel-chaos/src/main/docs/chaos.adoc | 7 +
.../component/chaos/dst/ChaosLifecycleDstTest.java | 92 +++++++
.../component/chaos/dst/ChaosLifecycleWorld.java | 268 +++++++++++++++++++++
.../camel/component/chaos/dst/FakeQueue.java | 163 +++++++++++++
.../src/test/resources/log4j2.properties | 9 +-
design/chaos.adoc | 3 +-
8 files changed, 572 insertions(+), 2 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc
index 84a54c419cd6..37621b180b9a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/chaos.adoc
@@ -426,6 +426,13 @@ scheduled executor of the `ExecutorServiceManager`. So in
a deterministic simula
in virtual time: a 10 minute stall takes milliseconds, and with
`camel.chaos.seed` set to the seed of the run, the same
seed gives the same report again, times included. Chaos does not depend on
`camel-test-dst`.
+`ChaosLifecycleDstTest` in the tests of camel-chaos is such a test, and an
example to copy: a route consumes a queue
+(a fake with a scheduled poll consumer) and calls a downstream system
asynchronously (`DstCamel.asyncStep`, as a world
+needs for a route that is suspended or stopped while an exchange is in a
downstream call). A `stall` or `lifecycle`
+rule, added while an exchange is in flight, takes the route down for up to 10
minutes while a `step=` rule fails the
+downstream call at a rate, and the test checks from the report that no message
is lost (an expectation), that the
+redeliveries add up to the step faults, and that the route was down for
exactly the duration.
+
The seed fixes the sequence of random draws of each rule
(`rate=<probability>`, the random header values), not which
exchange gets which draw: with several threads (concurrent consumers, parallel
processing) the threads decide that,
so two runs with the same seed can give different reports. With one thread and
the same exchanges in the same order,
diff --git a/components/camel-chaos/pom.xml b/components/camel-chaos/pom.xml
index 2088024c31be..bb1b30568a96 100644
--- a/components/camel-chaos/pom.xml
+++ b/components/camel-chaos/pom.xml
@@ -37,6 +37,8 @@
<label>tooling,testing</label>
<title>Chaos</title>
<supportLevel>Preview</supportLevel>
+ <!-- the DST test loads the byte code agent of camel-test-dst when the
JVM starts (see test-dst.adoc) -->
+
<camel.surefire.fork.additional-vmargs>"-javaagent:${net.bytebuddy:byte-buddy-agent:jar}"</camel.surefire.fork.additional-vmargs>
</properties>
<dependencies>
@@ -70,6 +72,12 @@
<artifactId>camel-file</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <!-- the deterministic simulation test that runs chaos rules in
virtual time (dst package) -->
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-dst</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<!-- the exception of the HTTP components, for the error fault
with status= -->
<groupId>org.apache.camel</groupId>
@@ -100,4 +108,21 @@
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <!-- sets the property net.bytebuddy:byte-buddy-agent:jar to
the path of the agent of the DST test -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>dst-agent-path</id>
+ <goals>
+ <goal>properties</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
</project>
diff --git a/components/camel-chaos/src/main/docs/chaos.adoc
b/components/camel-chaos/src/main/docs/chaos.adoc
index 84a54c419cd6..37621b180b9a 100644
--- a/components/camel-chaos/src/main/docs/chaos.adoc
+++ b/components/camel-chaos/src/main/docs/chaos.adoc
@@ -426,6 +426,13 @@ scheduled executor of the `ExecutorServiceManager`. So in
a deterministic simula
in virtual time: a 10 minute stall takes milliseconds, and with
`camel.chaos.seed` set to the seed of the run, the same
seed gives the same report again, times included. Chaos does not depend on
`camel-test-dst`.
+`ChaosLifecycleDstTest` in the tests of camel-chaos is such a test, and an
example to copy: a route consumes a queue
+(a fake with a scheduled poll consumer) and calls a downstream system
asynchronously (`DstCamel.asyncStep`, as a world
+needs for a route that is suspended or stopped while an exchange is in a
downstream call). A `stall` or `lifecycle`
+rule, added while an exchange is in flight, takes the route down for up to 10
minutes while a `step=` rule fails the
+downstream call at a rate, and the test checks from the report that no message
is lost (an expectation), that the
+redeliveries add up to the step faults, and that the route was down for
exactly the duration.
+
The seed fixes the sequence of random draws of each rule
(`rate=<probability>`, the random header values), not which
exchange gets which draw: with several threads (concurrent consumers, parallel
processing) the threads decide that,
so two runs with the same seed can give different reports. With one thread and
the same exchanges in the same order,
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/ChaosLifecycleDstTest.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/ChaosLifecycleDstTest.java
new file mode 100644
index 000000000000..13a1998e18cd
--- /dev/null
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/ChaosLifecycleDstTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.chaos.dst;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstInstrument;
+import org.apache.camel.test.dst.DstTest;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Chaos rules in virtual time: a route lifecycle rule and a step rule run
against a route in a camel-test-dst world
+ * ({@link ChaosLifecycleWorld}). A failing seed is replayed with {@code
-Ddst.seed=<seed>}; {@code -Ddst.seeds=5000}
+ * runs more seeds.
+ * <p/>
+ * camel-chaos reads the clock and sleeps only through {@code TimeSource}, and
its timed actions run on a scheduled
+ * executor of the {@code ExecutorServiceManager}, so it is instrumented here
only for the escape scan (strict mode).
+ * The route stop and suspend wait for the exchanges in flight in {@code
DefaultShutdownStrategy}, whose wait on its
+ * shutdown task is rewritten to run the loop.
+ */
+@DstInstrument(value = "org.apache.camel.component.chaos",
+ alsoRewrite = {
+ "org.apache.camel.util.StopWatch",
"org.apache.camel.impl.engine.DefaultShutdownStrategy",
+ "org.apache.camel.support.task" })
+class ChaosLifecycleDstTest {
+
+ @DstTest(seeds = 200)
+ void noMessageIsLostWhileARouteIsDownAndTheRedeliveriesAddUp(Dst dst)
throws Exception {
+ new ChaosLifecycleWorld(dst).run();
+ }
+
+ @Test
+ void theSameSeedGivesTheSameReportTimesIncluded() throws Exception {
+ for (long seed : List.of(3L, 11L, 42L)) {
+ String[] first = runOnce(seed);
+ String[] second = runOnce(seed);
+ assertThat(second[0]).as("report of seed %d",
seed).isEqualTo(first[0]);
+ assertThat(second[1]).as("trace of seed %d",
seed).isEqualTo(first[1]);
+ assertThat(first[0]).contains("\"downAt\"").contains("\"backAt\"");
+ }
+ }
+
+ @Test
+ void aTenMinuteStallTakesMilliseconds() throws Exception {
+ Dst dst = new Dst(7);
+ try {
+ dst.strict(true, List.of());
+ ChaosLifecycleWorld world = new ChaosLifecycleWorld(dst, "stall",
600_000L);
+ long start = System.nanoTime();
+ world.run();
+ long real = System.nanoTime() - start;
+
assertThat(world.report).contains("\"action\":\"suspend\"").contains("\"duration\":600000")
+ .contains("\"state\":\"back\"");
+ assertThat(dst.now()).isGreaterThanOrEqualTo(600_000L);
+ // 10 minutes of virtual time: far below a minute of real time
also on a busy machine
+ assertThat(real).isLessThan(TimeUnit.SECONDS.toNanos(30));
+ } finally {
+ dst.close();
+ }
+ }
+
+ /** One run of the world: its report and the hash of its trace. */
+ private static String[] runOnce(long seed) throws Exception {
+ Dst dst = new Dst(seed);
+ try {
+ dst.strict(true, List.of());
+ ChaosLifecycleWorld world = new ChaosLifecycleWorld(dst);
+ world.run();
+ return new String[] { world.report, dst.traceHash() };
+ } finally {
+ dst.close();
+ }
+ }
+}
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/ChaosLifecycleWorld.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/ChaosLifecycleWorld.java
new file mode 100644
index 000000000000..bfb0014bcd16
--- /dev/null
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/ChaosLifecycleWorld.java
@@ -0,0 +1,268 @@
+/*
+ * 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.chaos.dst;
+
+import java.util.List;
+
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.chaos.ChaosManager;
+import org.apache.camel.component.chaos.ChaosReport;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstCamel;
+import org.apache.camel.test.dst.DstExecutor;
+import org.apache.camel.test.dst.Invariants;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+
+/**
+ * A route under chaos rules in virtual time: a route consumes a {@link
FakeQueue} and calls a downstream system
+ * asynchronously ({@code DstCamel.asyncStep}, as the one-stack model of
camel-test-dst needs for a route that is
+ * suspended or stopped while an exchange is in a downstream call). Two chaos
rules, added through the public
+ * {@link ChaosManager} API, which {@code camel.chaos.rules} uses too:
+ * <ul>
+ * <li>a lifecycle rule ({@code stall}, or {@code lifecycle} with {@code
action=suspend}, {@code stop} or
+ * {@code remove}, for 1 to 10 minutes), added by an operator at a seeded time
once an exchange is in flight (so the
+ * suspend or stop waits for it), with an expectation that the route completes
every message it has not completed yet,
+ * and</li>
+ * <li>a step rule that fails the downstream call at a seeded rate, which the
error handler redelivers.</li>
+ * </ul>
+ * The seed of the run is {@code camel.chaos.seed}, and the seed picks the
rule, its time, the duration, the failure
+ * rate, the redeliveries and the messages: a burst before the rule, the
others after, one of them halfway through the
+ * time the route is down. The checks read the chaos report (no message is
lost: the expectation is met; the route
+ * completed every message once; the redeliveries are the step faults that did
not exhaust the error handler; the route
+ * was down for exactly the duration), the fake queue (every message deleted
once; none delivered while the route was
+ * down, the one published then only after) and the route controller (the
route is started).
+ * <p/>
+ * The thing under test is camel-chaos with the route lifecycle of camel-core,
not a component, so the world is not a
+ * {@code @DstWorld} of the catalog (which lists the worlds of components by
their scheme).
+ */
+final class ChaosLifecycleWorld {
+
+ static final String ROUTE = "orders";
+
+ private final Dst dst;
+ final String fault;
+ final long duration;
+ final int messages;
+ final String failureRate;
+ final int maxRedeliveries;
+ final long ruleAt;
+ final FakeQueue queue;
+
+ private int sendsLeft;
+ private String lifecycleId;
+ private long ruleAddedAt = -1;
+ private int inflightAtRule;
+ private DstExecutor producer;
+ // the report at the end of the run, as JSON text
+ String report;
+
+ ChaosLifecycleWorld(Dst dst) {
+ this(dst, null, -1);
+ }
+
+ /** A world with the given lifecycle fault and duration, or the seed's
pick for null and -1. */
+ ChaosLifecycleWorld(Dst dst, String fault, long duration) {
+ this.dst = dst;
+ this.fault = fault != null
+ ? fault : dst.pick(List.of("stall", "lifecycle
action=suspend", "lifecycle action=stop",
+ "lifecycle action=remove"));
+ this.duration = duration > 0 ? duration : dst.pick(List.of(60_000L,
300_000L, 600_000L));
+ this.messages = (int) dst.between(5, 30);
+ // a step rule from the start also makes the report count the route
from the start
+ this.failureRate = dst.pick(List.of("0.1", "0.3"));
+ this.maxRedeliveries = dst.pick(List.of(1, 2, 5));
+ this.ruleAt = dst.between(2_000, 10_000);
+ this.queue = new FakeQueue(dst);
+ dst.trace("world: fault=" + this.fault + " duration=" + this.duration
+ " messages=" + messages
+ + " failureRate=" + failureRate + " maxRedeliveries=" +
maxRedeliveries + " ruleAt=" + ruleAt);
+ }
+
+ void run() throws Exception {
+ DefaultCamelContext context = DstCamel.newContext(dst);
+ // the exchanges in flight finish while a route is suspended or
stopped: give them time (virtual)
+ context.getShutdownStrategy().setTimeout(300);
+
context.getPropertiesComponent().addInitialProperty("camel.chaos.enabled",
"true");
+
context.getPropertiesComponent().addInitialProperty("camel.chaos.seed",
String.valueOf(dst.seed()));
+
context.getPropertiesComponent().addInitialProperty("camel.chaos.maxDuration",
"1h");
+
context.getPropertiesComponent().addInitialProperty("camel.chaos.reportOnShutdown",
"false");
+ context.addComponent("queue", queue.component());
+ DstExecutor downstream = dst.executor("downstream", 1);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // redeliveries without a delay: a delay would sleep on the
thread of the exchange, and an exchange that
+ // sleeps below the route stop on the one stack of the run
could not finish (test-dst.adoc); with
+ // asyncDelayedRedelivery the delay runs on a thread of the
error handler outside the harness
+
errorHandler(defaultErrorHandler().maximumRedeliveries(maxRedeliveries).redeliveryDelay(0));
+
from("queue:orders?initialDelay=0&delay=500&maxMessagesPerPoll=5").routeId(ROUTE)
+ .process(DstCamel.asyncStep(dst, downstream, 100,
2_000, exchange -> {
+ })).id("downstream");
+ }
+ });
+ try {
+ context.start();
+ ChaosManager chaos = ChaosManager.get(context);
+ String stepRule = chaos.addRule("step=downstream error rate=" +
failureRate + " until=50m");
+ DstExecutor operator = dst.executor("operator", 1);
+ operator.schedule("add the lifecycle rule", ruleAt, () ->
addLifecycleRule(context, chaos, operator));
+ planSends();
+
+ long cap = 3_600_000L;
+ dst.runUntil(() -> sendsLeft == 0 && queue.drained() &&
lifecycleDone(chaos) || dst.now() > cap,
+ "all messages completed and the route back");
+ ChaosReport end = chaos.report();
+ report = end.toJson().toJson();
+ dst.trace("world: report " + report);
+ check(end, stepRule,
context.getRouteController().getRouteStatus(ROUTE));
+ } finally {
+ // after a run that drained the queue the stop waits for no
exchange; then the polls left are dropped
+ context.stop();
+ dst.dropPending();
+ }
+ }
+
+ /**
+ * The operator adds the lifecycle rule once an exchange is in flight in
the route (it retries every 50 ms, until
+ * the last message is published), and the producer publishes m0 halfway
through the time the route is down.
+ */
+ private void addLifecycleRule(DefaultCamelContext context, ChaosManager
chaos, DstExecutor operator) {
+ inflightAtRule = context.getInflightRepository().size(ROUTE);
+ if (inflightAtRule == 0 && dst.now() < ruleAt + duration + 60_000) {
+ operator.schedule("add the lifecycle rule", 50, () ->
addLifecycleRule(context, chaos, operator));
+ return;
+ }
+ ruleAddedAt = dst.now();
+ // the expectation counts the exchanges the route completes from now:
those of the messages not deleted yet
+ int remaining = messages - queue.deleted;
+ dst.trace("world: lifecycle rule with " + inflightAtRule + " exchanges
in flight, " + remaining + " to complete");
+ lifecycleId = chaos.addRule("route=" + ROUTE + " " + fault + "
duration=" + duration + "ms expect=" + ROUTE + ":"
+ + remaining + " within=50m");
+ producer.schedule("publish m0", duration / 2, () -> {
+ queue.publish("m0");
+ sendsLeft--;
+ });
+ }
+
+ private void planSends() {
+ producer = dst.executor("producer", 1);
+ // a burst before the rule, so an exchange is in flight when the route
goes down; m0 is published by the
+ // operator while the route is down; the others after the rule time,
while the route is down and after
+ int burst = (int) dst.between(2, Math.min(6, messages - 1));
+ long span = duration + 60_000;
+ sendsLeft++;
+ for (int i = 1; i < messages; i++) {
+ String id = "m" + i;
+ sendsLeft++;
+ long at = i <= burst ? ruleAt - dst.between(1, 1_500) : ruleAt +
dst.between(1, span);
+ producer.schedule("publish " + id, at, () -> {
+ queue.publish(id);
+ sendsLeft--;
+ });
+ }
+ }
+
+ private boolean lifecycleDone(ChaosManager chaos) {
+ JsonObject l = lifecycle(chaos.report().toJson());
+ return l != null && List.of("back", "skipped", "left", "failed",
"cancelled").contains(l.getString("state"));
+ }
+
+ private JsonObject rule(JsonObject report, String id) {
+ for (Object o : (JsonArray) report.get("rules")) {
+ JsonObject rule = (JsonObject) o;
+ if (rule.getString("id").equals(id)) {
+ return rule;
+ }
+ }
+ return null;
+ }
+
+ private JsonObject lifecycle(JsonObject report) {
+ JsonObject rule = lifecycleId != null ? rule(report, lifecycleId) :
null;
+ return rule != null ? (JsonObject) rule.get("lifecycle") : null;
+ }
+
+ private void check(ChaosReport end, String stepRule, ServiceStatus status)
{
+ JsonObject json = end.toJson();
+ JsonObject routes = null;
+ for (Object o : (JsonArray) json.get("routes")) {
+ if (ROUTE.equals(((JsonObject) o).getString("route"))) {
+ routes = (JsonObject) o;
+ }
+ }
+ JsonObject counts = routes;
+ JsonObject lifecycle = lifecycle(json);
+ String config = " (fault=" + fault + " duration=" + duration + "
messages=" + messages + " failureRate="
+ + failureRate + " maxRedeliveries=" + maxRedeliveries
+ ")";
+ dst.check(
+ Invariants.of("no message is lost: the expectation of the
lifecycle rule is met, the verdict PASS",
+ () -> ChaosReport.PASS.equals(end.getVerdict()) &&
end.isFinal() && end.getLost() == 0
+ && "met".equals(((JsonObject) rule(json,
lifecycleId).get("expectation"))
+ .getString("outcome"))
+ ? null : end.toText() +
config),
+ Invariants.of("the route completed every message once",
+ () -> counts != null && counts.getLong("completed") ==
messages
+ ? null : "routes " + counts + config),
+ Invariants.of("the redeliveries are the step faults that did
not exhaust the error handler",
+ () -> {
+ if (counts == null) {
+ return "no route entry in the report" + config;
+ }
+ long fired = ((JsonObject) rule(json,
stepRule).get("fired")).getLong("step");
+ long expected = fired - counts.getLong("failed");
+ return counts.getLong("redeliveries") == expected
+ && counts.getLong("failed") ==
queue.failedBack
+ ? null
+ : "fired " + fired + ", routes " +
counts + ", failed back to the queue "
+ + queue.failedBack + config;
+ }),
+ Invariants.of("the route was down for the duration, and is
back",
+ () -> lifecycle != null &&
"back".equals(lifecycle.getString("state"))
+ && lifecycle.getLong("backAt") -
lifecycle.getLong("downAt") == duration
+ ? null : "lifecycle " + lifecycle +
config),
+ Invariants.of("the route took no message while it was down",
+ () -> {
+ if (lifecycle == null ||
!lifecycle.containsKey("backAt")) {
+ return "the route did not come back: lifecycle
" + lifecycle + config;
+ }
+ long from = ruleAddedAt +
lifecycle.getLong("downAt");
+ long to = ruleAddedAt +
lifecycle.getLong("backAt");
+ List<Long> during =
queue.deliveries.stream().filter(t -> t > from && t < to).toList();
+ return during.isEmpty()
+ ? null : "deliveries at " + during + "
while down from " + from + " to " + to + config;
+ }),
+ Invariants.of("the message published while the route was down
was delivered after it came back",
+ () -> {
+ if (lifecycle == null ||
!lifecycle.containsKey("backAt")) {
+ return "the route did not come back: lifecycle
" + lifecycle + config;
+ }
+ Long first = queue.firstDelivery.get("m0");
+ long back = ruleAddedAt +
lifecycle.getLong("backAt");
+ return first != null && first >= back
+ ? null : "m0 first delivered at " + first
+ ", back at " + back + config;
+ }),
+ Invariants.of("the queue deleted every message once, and is
drained",
+ () -> queue.drained() && queue.deleted == messages
+ ? null : queue.size() + " messages left, " +
queue.deleted + " deleted" + config),
+ Invariants.of("an exchange was in flight when the lifecycle
rule was added",
+ () -> inflightAtRule > 0 ? null : "no exchange in
flight" + config),
+ Invariants.of("the route is started at the end",
+ () -> status == ServiceStatus.Started ? null : "route
" + ROUTE + " is " + status + config));
+ }
+}
diff --git
a/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/FakeQueue.java
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/FakeQueue.java
new file mode 100644
index 000000000000..77910ef9e79f
--- /dev/null
+++
b/components/camel-chaos/src/test/java/org/apache/camel/component/chaos/dst/FakeQueue.java
@@ -0,0 +1,163 @@
+/*
+ * 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.chaos.dst;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.support.ScheduledPollConsumer;
+import org.apache.camel.support.ScheduledPollEndpoint;
+import org.apache.camel.test.dst.Dst;
+
+/**
+ * A queue that keeps a message until the route completes it, as a broker with
client acknowledgement does, and its
+ * consumer: a {@link ScheduledPollConsumer}, the base of the file, FTP and
AWS SQS consumers, so it is suspended,
+ * stopped and started as theirs are.
+ * <p/>
+ * The queue is the external system of the world; it lives outside the
endpoint, so a route that is removed and added
+ * back finds the same messages. A poll takes up to {@code maxMessagesPerPoll}
visible messages and hands each to the
+ * route without waiting for it (the asynchronous processor); while the route
has a message, the message is not visible.
+ * When the route is done with the exchange, a completed message is deleted
and a failed one becomes visible again, to
+ * be polled later. There are no latencies, faults or duplicate deliveries of
its own: the faults of the world are the
+ * chaos rules.
+ */
+final class FakeQueue {
+
+ private final Dst dst;
+ // the messages by id, in the order they were published: whether each is
in flight
+ private final Map<String, Boolean> messages = new LinkedHashMap<>();
+ // the virtual times (dst.now()) of the deliveries to the route, for the
invariants
+ final List<Long> deliveries = new ArrayList<>();
+ // the virtual time of the first delivery of each message
+ final Map<String, Long> firstDelivery = new LinkedHashMap<>();
+ int deleted;
+ int failedBack;
+
+ FakeQueue(Dst dst) {
+ this.dst = dst;
+ }
+
+ void publish(String id) {
+ messages.put(id, false);
+ }
+
+ boolean drained() {
+ return messages.isEmpty();
+ }
+
+ int size() {
+ return messages.size();
+ }
+
+ private List<String> receive(int max) {
+ List<String> answer = new ArrayList<>();
+ for (Map.Entry<String, Boolean> e : messages.entrySet()) {
+ if (answer.size() == max) {
+ break;
+ }
+ if (!e.getValue()) {
+ e.setValue(true);
+ answer.add(e.getKey());
+ }
+ }
+ return answer;
+ }
+
+ private void done(String id, boolean completed) {
+ if (completed) {
+ messages.remove(id);
+ deleted++;
+ dst.trace("queue: " + id + " deleted");
+ } else {
+ messages.put(id, false);
+ failedBack++;
+ dst.trace("queue: " + id + " failed, visible again");
+ }
+ }
+
+ DefaultComponent component() {
+ return new DefaultComponent() {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters)
+ throws Exception {
+ QueueEndpoint endpoint = new QueueEndpoint(uri, this);
+ setProperties(endpoint, parameters);
+ return endpoint;
+ }
+ };
+ }
+
+ final class QueueEndpoint extends ScheduledPollEndpoint {
+
+ private int maxMessagesPerPoll = 5;
+
+ QueueEndpoint(String uri, DefaultComponent component) {
+ super(uri, component);
+ }
+
+ public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
+ this.maxMessagesPerPoll = maxMessagesPerPoll;
+ }
+
+ @Override
+ public Producer createProducer() {
+ throw new UnsupportedOperationException("The fake queue of the
world has no producer");
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ QueueConsumer consumer = new QueueConsumer(this, processor,
maxMessagesPerPoll);
+ configureConsumer(consumer);
+ return consumer;
+ }
+ }
+
+ final class QueueConsumer extends ScheduledPollConsumer {
+
+ private final int max;
+
+ QueueConsumer(Endpoint endpoint, Processor processor, int max) {
+ super(endpoint, processor);
+ this.max = max;
+ }
+
+ @Override
+ protected int poll() {
+ List<String> batch = receive(max);
+ for (String id : batch) {
+ Exchange exchange = createExchange(false);
+ exchange.getMessage().setBody(id);
+ deliveries.add(dst.now());
+ firstDelivery.putIfAbsent(id, dst.now());
+ dst.trace("queue: " + id + " delivered");
+ getAsyncProcessor().process(exchange, doneSync -> {
+ done(id, !exchange.isFailed() &&
!exchange.isRollbackOnly());
+ releaseExchange(exchange, false);
+ });
+ }
+ return batch.size();
+ }
+ }
+}
diff --git a/components/camel-chaos/src/test/resources/log4j2.properties
b/components/camel-chaos/src/test/resources/log4j2.properties
index 8ff8d5d6a66b..79841176c66b 100644
--- a/components/camel-chaos/src/test/resources/log4j2.properties
+++ b/components/camel-chaos/src/test/resources/log4j2.properties
@@ -15,9 +15,16 @@
## limitations under the License.
## ---------------------------------------------------------------------------
-appender.file.type = File
+# capped: the deterministic simulation test (dst package) runs many seeds
+appender.file.type = RollingFile
appender.file.name = file
appender.file.fileName = target/camel-chaos-test.log
+appender.file.filePattern = target/camel-chaos-test.%i.log
+appender.file.policies.type = Policies
+appender.file.policies.size.type = SizeBasedTriggeringPolicy
+appender.file.policies.size.size = 50MB
+appender.file.strategy.type = DefaultRolloverStrategy
+appender.file.strategy.max = 1
appender.file.layout.type = PatternLayout
appender.file.layout.pattern = %d %-5p %c{1} - %m %n
appender.out.type = Console
diff --git a/design/chaos.adoc b/design/chaos.adoc
index e97071b01a00..cc7bbc81b435 100644
--- a/design/chaos.adoc
+++ b/design/chaos.adoc
@@ -429,7 +429,8 @@ Each step is its own pull request after this document:
. camel-core: the step hook in the error handler of each step (CAMEL-26338),
then the faults at steps. Done.
. Route lifecycle faults. Done.
. The MCP tool group. Done.
-. A DST world that runs chaos rules in virtual time, as an example and a
regression test.
+. A DST world that runs chaos rules in virtual time, as an example and a
regression test. Done.
+ (`ChaosLifecycleDstTest` in the tests of camel-chaos).
== Alternatives considered
