allthingssecurity commented on code in PR #28067: URL: https://github.com/apache/camel/pull/28067#discussion_r4311122826
########## components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/DstClock.java: ########## @@ -0,0 +1,252 @@ +/* + * 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.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.OffsetDateTime; +import java.time.OffsetTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.time.ZonedDateTime; +import java.util.concurrent.TimeUnit; + +/** + * 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> + * 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}, + * {@code LocalTime} and {@code OffsetTime}. A {@link Clock} returned here reads the virtual time of whichever run is + * active when it is read, so a clock kept in a field works across runs. + * <p> + * Sleeps are rewritten too ({@code Thread.sleep} and {@code TimeUnit.sleep}): on the loop thread a sleep is a virtual + * wait ({@link Dst#block(long, String)}), so the sleeping task keeps its thread and the other tasks run meanwhile. + */ +public final class DstClock { + + private static final long NANO_OFFSET = 1_000_000_000_000L; + + private DstClock() { + } + + private static Dst read() { + Dst dst = Dst.active(); + if (dst != null && !dst.isLoopThread()) { + DstEscapes.recordRuntime("virtual clock read from a thread outside the harness: " Review Comment: Right, thanks. Done in 0b1c0ebc3866: a virtual clock read and all four sleep overloads off the loop now go through `DstEscapes.escaped(dst, ...)`, like `reached()`, so strict mode fails the seed and `allowEscapes` applies; the read still returns the virtual time and the sleep still sleeps in real time. Four new self-tests in `DstStrictModeTest` (strict read, strict sleep, allowed, non-strict recorded). The SQS, Event Hubs and Slack worlds run in strict mode and reach none of these, so none needed an `allowEscapes` entry. _Claude Code on behalf of allthingssecurity_ ########## components/camel-test/camel-test-dst/src/main/java/org/apache/camel/test/dst/Dst.java: ########## @@ -0,0 +1,643 @@ +/* + * 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.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HexFormat; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.Callable; +import java.util.function.BooleanSupplier; + +/** + * 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 + * a step trace. + * <p> + * Rules: + * <ul> + * <li>Only the thread that created the run (the loop thread, the JUnit test thread) runs tasks. Tasks run one at a time + * and are never preempted: the unit of interleaving is a task.</li> + * <li>Virtual time only moves when nothing is runnable at the current time; it then jumps to the next due task.</li> + * <li>When several tasks are runnable, the seeded random source picks one. A single-thread executor runs its tasks in + * FIFO order, an executor with n threads offers its first n runnable tasks.</li> + * <li>A blocking call of a fake external system is a yield point ({@link #block(long, String)}): the calling task's + * thread stays busy and the loop runs other tasks (nested) until the call returns.</li> + * </ul> + * The same seed gives the same sequence of steps, so a failing seed is replayed exactly. + */ +public final class Dst { + + /** Virtual time 0 is this wall clock instant, so code that prints times sees plausible values. */ + public static final long EPOCH_MILLIS = 1_700_000_000_000L; + + private static final int MAX_NESTING = 64; + /** An untimed wait that has not ended after this much virtual time is reported as a hang. */ + private static final long MAX_WAIT_MILLIS = 24L * 3600 * 1000; + private static final int MAX_TRACE_LINES = 200_000; + + private static volatile Dst active; + + private final long seed; + private final Random random; + private final Thread loopThread; + private final long maxSteps; + private final List<DstTask<?>> pending = new ArrayList<>(); + private final List<String> trace = new ArrayList<>(); + private final ArrayDeque<String> traceTail = new ArrayDeque<>(); + private final List<DstTask<?>> running = new ArrayList<>(); + private final DstExecutor wakeups; + private long now; + private long sequence; + private long steps; + private int nesting; + private boolean traceTruncated; + private boolean closed; + private volatile long progress; + private volatile long watchdogMillis = Long.getLong("dst.watchdogMillis", 10_000); + private volatile String stall; + private boolean stallThrown; + private volatile boolean strict; + private volatile List<String> allowedEscapes = List.of(); + private volatile String escape; + private boolean escapeThrown; + + public Dst(long seed) { + this(seed, 1_000_000); + } + + public Dst(long seed, long maxSteps) { + this.seed = seed; + this.random = new Random(scramble(seed)); + this.loopThread = Thread.currentThread(); + this.maxSteps = maxSteps; + this.wakeups = new DstExecutor(this, "wakeup", Integer.MAX_VALUE); + active = this; Review Comment: Done in 0b1c0ebc3866 with your check (under a lock, and `close()` clears `active` under the same lock): a second run while one is open throws `DST: another run (seed N) is still active; one run per JVM at a time`. `DstExtension` also refuses to run when `junit.jupiter.execution.parallel.enabled` is true. New `DstOneRunTest`. Javadoc of `active()` fixed: the run is JVM-wide. _Claude Code on behalf of allthingssecurity_ ########## components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc: ########## @@ -0,0 +1,406 @@ += Test DST Component +:doctitle: Test DST +:shortname: test-dst +:artifactid: camel-test-dst +:description: Deterministic simulation testing for Camel consumers: seeded virtual-time scheduler, fault injection, invariants and seed replay +:since: 4.24 +:supportlevel: Preview +:tabs-sync-option: + +*Since Camel {since}* + +The `camel-test-dst` module runs the real code of a Camel consumer under a deterministic, seeded, virtual-time +scheduler. Only the external system is faked. A test runs thousands of seeds in seconds, and a failing seed replays +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. +* 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. + +== A test + +[source,java] +---- +@DstInstrument("org.apache.camel.component.aws2.sqs") +class SqsVisibilityDstTest { + + @DstTest(seeds = 1000) + void noMessageIsDeliveredAgainWhileTheConsumerHoldsIt(Dst dst) throws Exception { + new SqsWorld(dst).run(); + } +} +---- + +The world (`SqsWorld`) builds a `CamelContext` with `DstCamel.newContext(dst)`, binds the fake client, starts the +route, drives the run with `dst.runUntil(...)` and `dst.runFor(...)`, and checks the invariants with `dst.check(...)`. +The test and its world live in a package of their own (for example `org.apache.camel.component.aws2.sqs.dst`); that +package is never rewritten. + +Three 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-azure-eventhubs |`EventHubsCheckpointDstTest` (checkpointing by batch size and timeout) |CAMEL-25683 +|camel-slack |`SlackSuspendDstTest` (route suspend in the middle of a poll) |CAMEL-25684 +|=== + +== What the harness controls + +[cols="1,3"] +|=== +|Thread pools |Every pool and scheduled executor created by Camel or the component through the +`ExecutorServiceManager`. A single-thread executor keeps its FIFO order; an executor with n threads offers its first n +runnable tasks to the seed. +|Time reads |`System.currentTimeMillis()`, `System.nanoTime()`, `new Date()`, `Instant.now()`, `Clock.systemUTC()`, +`Clock.systemDefaultZone()`, `Clock.system(ZoneId)` and `now()` / `now(ZoneId)` of `LocalDateTime`, `ZonedDateTime`, +`OffsetDateTime`, `LocalDate`, `LocalTime` and `OffsetTime`, in the packages of `@DstInstrument`. +|Sleeps |`Thread.sleep` and `TimeUnit.sleep`: on the loop thread a sleep is a virtual wait. The sleeping task keeps +its thread and the other tasks run meanwhile. +|Blocking waits |`CompletableFuture.get/join`, `Future.get`, `CountDownLatch.await`, `Semaphore.acquire/tryAcquire`, +`DelayQueue` and `BlockingQueue` `take/poll`: the loop runs until the wait can end, and a timed wait gives up in +virtual time. A wait that no task of the run can end fails the seed at once and names the call site. +|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. +|=== + +`@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. + +=== Escapes and strict mode + +Before the first seed, the harness scans the packages under test and writes every *escape point* to +`target/dst/<Test>.escapes.txt`: own threads and executors, the common fork-join pool, Reactor schedulers, waits it +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, the seed fails with the escape as its cause. `@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. + +=== The one-stack model + +All tasks run on one thread, the test thread. A task that blocks (a call of the fake with a latency, a sleep, a wait) +stays on the stack while the loop runs the other tasks *nested* above it. So a blocked task can resume only after +every task that blocked after it has resumed. Three consequences: + +* Timing: see the rule in the guide below. +* A timed wait notices its timeout only when the tasks nested above it return. If the awaited task completes after the + deadline that way, the caller still gets the `TimeoutException`, but the task's side effects have already happened. +* A route stop or suspend that runs while an exchange is blocked in a synchronous call below it on the stack cannot see + that exchange finish. The Slack world therefore copies the suspend sequence of `DefaultShutdownStrategy` with a + short polling wait instead of calling `suspendRoute`. With an asynchronous downstream call + (`DstCamel.asyncStep`) a real `stopRoute` works. + +Code that uses its own threads or a third-party scheduler (the Azure SDKs on Reactor) is outside the harness; fake the +client at a level above those threads. + +== Writing a fake world: a guide for AI agents and humans + +A world is the component's real consumer, a fake of the external system, a load (producers, operators), seeded faults +and invariants. This guide uses the SQS world of camel-aws2-sqs (`FakeSqs`, `SqsWorld`, about 420 lines) as the worked +example. Follow the steps in order. + +=== 1. Find where the client is swapped + +Look for the place where the component gets its client: + +* an endpoint option that takes a client instance, bound in the registry. camel-aws2-sqs has `amazonSQSClient`: ++ +[source,java] +---- +context.getRegistry().bind("sqs", fakeSqs); // FakeSqs implements SqsClient +from("aws2-sqs://dst?amazonSQSClient=#sqs&queueUrl=RAW(" + FakeSqs.QUEUE_URL + ")&extendMessageVisibility=true" + + "&visibilityTimeout=" + visibilityTimeout + "&maxMessagesPerPoll=" + maxMessagesPerPoll + + "&concurrentConsumers=" + concurrentConsumers + "&initialDelay=100&delay=" + delay) +---- +* else a static factory, replaced with Mockito's `mockStatic` for the duration of the run (the Event Hubs world mocks + `EventHubsClientFactory`, the Slack world mocks `Slack.getInstance`). + +Fake the client at the level where the component calls it synchronously. If the client runs its own threads (Reactor, +Netty, an HTTP client pool), fake it above them. + +=== 2. Run the escape scan before writing anything + +Write the smallest test with `@DstInstrument("<component package>")` and one seed, run it, and read +`target/dst/<Test>.escapes.txt`. Every line is a place where the component leaves the harness. For each one, decide: + +* *fix* it in the harness (the time reads, sleeps and waits listed above are already handled), +* *avoid* it with the configuration of the world (the first SQS world kept `maxMessagesPerPoll` at 10 or less because + more joined several requests with `CompletableFuture.get()`; the harness now runs the loop for that wait), or +* *document* it, in the world's Javadoc, and allow it in strict mode only if it cannot change the result. + +The SQS scan flagged `Instant.now()` in the visibility extender. Before it was rewritten, 271 of 500 seeds on `main` +failed falsely: the extender computed its deadline from real time while the fake counted in virtual time. + +=== 3. Write the fake from the service's documentation + +Write the fake from the API reference of the service, not from the component's code: a fake copied from the consumer +agrees with the consumer's bugs. Quote the rule next to the code that implements it, name the page, and mark what is +assumed: + +[source,java] +---- +// [R] "Valid values: 1 to 10. Default: 1." and "Amazon SQS never returns more messages than this value +// (however, fewer messages might be returned)" +int max = request.maxNumberOfMessages() != null ? request.maxNumberOfMessages() : 1; +... +// [D] "If you use an old ReceiptHandle, the request will succeed, but the message might not be deleted." +if (dst.chance(0.5)) { + delete(m, "deleted (old handle)"); +} +... +// [C] "ReceiptHandleIsInvalid: The specified receipt handle isn't valid." (assumed for a deleted +// message and for the handle of an earlier receive; the page does not spell these cases out) +---- + +Where the documentation allows several behaviours ("might", "fewer messages might be returned"), let the seed choose. +List what is not modelled in the class Javadoc (`FakeSqs`: long polling, FIFO queues, the in-flight limit, retention, +dead-letter queues). + +=== 4. The one-stack timing rule + +A blocked task resumes only after the tasks that blocked after it. A call of the fake that blocks for its latency can +therefore return later than its latency, and anything it decides at the start can be stale at the end. The rule: + +* *reads take effect at the end of their latency*: `FakeSqs.receiveMessage` blocks first, then picks the visible + messages and starts their visibility timeout; +* *writes take effect at the start*: `deleteMessage` deletes, then blocks for its latency; +* *background tasks do not block*: the visibility extender of the consumer runs on a scheduled executor, so + `changeMessageVisibilityBatch` answers without a latency. If it blocked, a delayed resume could make the extension + arrive after the timeout it extends, which is not what SQS does. + +Write this down in the fake's Javadoc, as `FakeSqs` does. + +=== 5. Let the seed pick the configuration + +The seed picks the endpoint configuration as well as the timing, at the start of the run, and the world writes it to +the trace: + +[source,java] +---- +this.visibilityTimeout = dst.pick(List.of(2, 3, 5, 10)); +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)); +dst.trace("world: visibilityTimeout=" + visibilityTimeout + " maxMessagesPerPoll=" + maxMessagesPerPoll + ...); +---- + +This found CAMEL-25737: with `visibilityTimeout=1` the extender is always late. It is left out of the default set while +that issue is open, and `-Ddst.sqs.visibilityTimeouts=1` puts it back. The seed of a run is scrambled (SplitMix64), so +consecutive seeds draw unrelated configurations. + +=== 6. Seeded faults, a drain phase, and fault-excuse records + +Inject the faults the service documents (failed calls, slow answers, lost requests) with `dst.chance(...)`, at a rate +that still lets work get done (the SQS world: 5 % failed receives and deletes, 2 % failed visibility changes, 15 % slow +processing). Then switch the faults off and let the system drain: + +[source,java] +---- +dst.runUntil(() -> sendsLeft == 0, "all sends"); +dst.runFor(dst.between(0, 3L * visibilityTimeout * 1000)); +// faults cleared: the queue must drain +faults = false; +sqs.faults = false; +long deadline = dst.now() + 20L * visibilityTimeout * 1000 + 30_000; +while (!sqs.allDeleted() && dst.now() < deadline) { + dst.runFor(100); +} +---- + +Some faults legitimately cause what the invariant forbids. Record which fault excuses which violation, and only those: +a failed `ChangeMessageVisibilityBatch` lets SQS deliver the messages of that batch again, so `FakeSqs` reports their +ids (`onExtensionFault`) and the world does not count a redelivery of those messages. + +=== 7. The invariant comes from the component's contract + +Write the property the component promises, not what the code happens to do. The visibility extender promises that a +message is not delivered again while the consumer holds it. The world records what happened in a `DstLedger` and +checks: + +[source,java] +---- +dst.check( + Invariants.of("no message is delivered again while the consumer holds it", + () -> redeliveredWhileHeld.isEmpty() ? null : redeliveredWhileHeld.toString()), + Invariants.everyOfferedRoutedAtLeastOnce(ledger, "sent", "completed"), + Invariants.of("the queue is drained once the faults cleared", + () -> sqs.allDeleted() ? null : "not deleted: " + sqs.notDeleted())); +---- + +Start from the reusable `Invariants` (`everyOfferedRoutedAtLeastOnce`, `noDuplicates`, +`lastCompletedAcknowledgedWhenQuiet`, `acknowledgedWithin`, `routeStarted`) and add the component's own with +`Invariants.of(name, check)`. A violation message carries the configuration, so the failure report is readable on its +own. + +=== 8. The self-check + +A world is only evidence when it can fail. Before trusting a pass on `main`: + +. *Fail on the code before a past fix.* Find a fixed bug of the same behaviour with `git log` on the component + (the SQS world: CAMEL-25700, commit d09593b4a584). Revert the fix locally and run the world: it must fail (1397 of 5000 seeds), and only + in the configurations the fix describes (for CAMEL-25700: `maxMessagesPerPoll >= 2` and `concurrentConsumers >= 2`). +. *Pass on main*, with thousands of seeds (`-Ddst.seeds=5000`). +. *Replay*: run a failing seed twice with `-Ddst.seed=<seed>`. The trace files in `target/dst/` must have the same + SHA-256 (it is on their second line). + +If the world cannot fail on the old code, its fake, its load or its invariant is too weak. + +=== 9. Triage before reporting + +A failing seed is not yet a bug. Before reporting one: + +. Group the failing seeds by configuration (the `world:` line of each trace, or the violation message). A failure + that appears only in one configuration points at that configuration (CAMEL-25737 was all `visibilityTimeout=1`). +. Read one trace from the start: what the world chose, then each step with its virtual time and executor. +. Tell three causes apart: +** a *harness artifact*: an escape (check the summary), the one-stack rule broken by the fake, a time read that is + not rewritten (before the `Instant.now()` rewrite, half of the SQS seeds failed this way); +** a *fake mistake*: the fake does something the service documentation does not allow; +** a *real bug*: the consumer breaks its contract against a fake that follows the documentation. +. Reproduce a real bug outside DST too, with a plain unit test, before filing it. + +=== Pitfalls seen when agents wrote worlds + +Agents wrote nine worlds with this guide. These mistakes came up: + +* *Timestamps at 0.* Give the times of the fake virtual wall clock time (`Dst.EPOCH_MILLIS + dst.now()`, or + `dst.currentTimeMillis()`), not 0. A Service Bus fake with lock times in 1970 made the lock renewer renew on every run, + so the world could not fail. The self-check (fail on the code before the fix) caught it. +* *No end to the run.* A consumer that loops forever needs an unwind at the end of the run, or the run never ends: + cancel its task through the task registry, or stop the route at the end of the drain phase. +* *A fake that blocks under a lock.* A call of the fake made while the consumer holds a lock must not block: with the + one-stack rule, the tasks nested above it run while the lock is held. +* *A violation without its excuse.* Every invariant needs the record of which injected faults excuse a violation of it. + Without it, a message dead-lettered after injected acknowledgement faults showed up as a false failure. +* *Old traces.* Clear `target/dst` between runs, or the traces of an earlier run end up in the grouping of the failing + seeds. +* *A loud test log.* A consumer looping under a buggy fake wrote a 31 GB test log and filled the disk. Set the + component to `WARN` in the test `log4j2` configuration, or cap the log. +* *A fixed configuration.* Let the seed pick the endpoint configuration, edge values included. That is how + CAMEL-25737 (`visibilityTimeout=1`) and CAMEL-25741 (aws2-ddbstream, found with a fake DynamoDB Streams world) were + found. + +=== Checklist + +* [ ] The client is swapped through an endpoint option or a static mock, at a synchronous level. +* [ ] The escape scan was read; every escape is fixed, avoided or documented; the test runs in strict mode. +* [ ] Every behaviour of the fake quotes the documentation; assumptions are marked; what is not modelled is listed. +* [ ] Reads take effect at the end of their latency, writes at the start; background tasks do not block. +* [ ] The seed picks the configuration, edge values included, and the world traces it. +* [ ] The fake's timestamps are virtual wall clock time (`Dst.EPOCH_MILLIS` based), not 0. +* [ ] A consumer that loops forever is unwound at the end of the run. +* [ ] Faults are seeded, switched off for a drain phase, and recorded where they excuse a violation. +* [ ] The component logs at `WARN` in the test log configuration, and `target/dst` is cleared between runs. +* [ ] The invariant states the component's contract. +* [ ] The world fails on the code before a past fix, passes on `main`, and a seed replays with the same SHA-256. + +== Running the tests + +The worlds are tests of their components, in a `dst` package. Build the harness in the same reactor as the component +(`-pl` both), so that the component's tests see it. `-Dsurefire.rerunFailingTestsCount=0` stops surefire from running +each failing seed again. + +[source,bash] +---- +DST=components/camel-test/camel-test-dst +SQS=components/camel-aws/camel-aws2-sqs +EH=components/camel-azure/camel-azure-eventhubs +SLACK=components/camel-slack +MVN="./mvnw -B -Papache-snapshots test -Dsurefire.failIfNoSpecifiedTests=false -Dsurefire.rerunFailingTestsCount=0" + +# the harness self-tests +$MVN -pl $DST -Dtest='Dst*Test' + +# the worlds, 1000 seeds each, then 5000 seeds without stopping at the first failing one +$MVN -pl $DST,$SQS,$EH,$SLACK -Dtest='*DstTest' +$MVN -pl $DST,$SQS,$EH,$SLACK -Dtest='*DstTest' -Ddst.seeds=5000 -Ddst.maxFailures=100000 + +# the self-check: revert the main code of a past fix locally (do not commit), run the world, restore +git show d09593b4a584 -- $SQS/src/main | git apply -R # CAMEL-25700 +$MVN -pl $DST,$SQS -Dtest=SqsVisibilityDstTest -Ddst.seeds=5000 -Ddst.maxFailures=100000 +git checkout -- $SQS/src/main + +# replay one seed twice and compare the traces +$MVN -pl $DST,$SQS -Dtest=SqsVisibilityDstTest -Ddst.seed=8 +cp $SQS/target/dst/SqsVisibilityDstTest.*/seed-8.trace /tmp/seed-8.first +$MVN -pl $DST,$SQS -Dtest=SqsVisibilityDstTest -Ddst.seed=8 +diff /tmp/seed-8.first $SQS/target/dst/SqsVisibilityDstTest.*/seed-8.trace && echo identical +---- + +The past fixes of the other two worlds are 83aaa80754f9 (CAMEL-25683, Event Hubs) and 68329f0919bc (CAMEL-25684, +Slack). Two system properties belong to a world rather than to the harness: `dst.sqs.visibilityTimeouts` (the +visibility timeouts the SQS world picks from, default `2,3,5,10`; `1` shows CAMEL-25737) and `dst.replayCheckSeed` (the +seed of the in-JVM replay test of the Event Hubs world, default 1). + +== Limits + +* Interleaving is per task: the harness does not preempt a task in the middle of a method. +* The one-stack model above: a blocked task resumes only after the tasks that blocked after it. +* 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 is a static field. Review Comment: Added under Limits in 0b1c0ebc3866 (module doc and the catalog copy, still identical): a world must not go into a module with `camel.surefire.parallel=true` (and the harness now refuses it), and the `@DstInstrument` packages, including the test classes in the same package, stay rewritten for the rest of the surefire fork, reading the real clock outside a run. _Claude Code on behalf of allthingssecurity_ ########## components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/dst/EventHubsWorld.java: ########## @@ -0,0 +1,310 @@ +/* + * 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.azure.eventhubs.dst; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; + +import com.azure.messaging.eventhubs.CheckpointStore; +import com.azure.messaging.eventhubs.EventData; +import com.azure.messaging.eventhubs.EventProcessorClient; +import com.azure.messaging.eventhubs.models.Checkpoint; +import com.azure.messaging.eventhubs.models.EventContext; +import com.azure.messaging.eventhubs.models.PartitionContext; +import com.azure.messaging.eventhubs.models.PartitionOwnership; +import org.apache.camel.AsyncCallback; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.azure.eventhubs.EventHubsConsumer; +import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.AsyncProcessorSupport; +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.DstLedger; +import org.apache.camel.test.dst.Invariant; +import org.apache.camel.test.dst.Invariants; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import static org.mockito.ArgumentMatchers.any; + +/** + * The world around a camel-azure-eventhubs consumer: a fake Event Hubs SDK that delivers events per partition (bursts + * and quiet periods chosen by the seed), an asynchronous downstream step with latency and failures, and a fake + * checkpoint store whose answers are delayed, slow or failing. The consumer, the route and Camel's scheduler are real; + * every thread pool runs on the DST loop. + */ +final class EventHubsWorld { + + static final String ROUTE = "eh"; + + /** Probe counters over all seeds of the JVM (davsclaus' corner case on #27922 and the other causes). */ + static final Map<String, Integer> PROBES = new TreeMap<>(); + static final Map<String, Set<Long>> PROBE_SEEDS = new TreeMap<>(); + + private final Dst dst; + private final DstLedger ledger; + private final Set<String> probes = new TreeSet<>(); + private final int batchSize; + private final long timeout; + private final int partitions; + private final long maxStoreLatency; + private Consumer<EventContext> processEvent; + private EventHubsConsumer consumer; + private int inflight; + private int arrivalsLeft; + private final Map<String, Integer> sizeCheckpoints = new TreeMap<>(); + private final DstExecutor store; + + private final boolean countProbes; + + EventHubsWorld(Dst dst) { + this(dst, true); + } + + EventHubsWorld(Dst dst, boolean countProbes) { + this.dst = dst; + this.countProbes = countProbes; + this.ledger = new DstLedger(dst); + this.batchSize = dst.pick(List.of(2, 3, 5, 10)); + this.timeout = dst.pick(List.of(200L, 500L, 1000L)); + this.partitions = dst.pick(List.of(1, 2)); + this.maxStoreLatency = 3 * timeout; + this.store = dst.executor("checkpoint-store", Integer.MAX_VALUE); + dst.trace("world: checkpointBatchSize=" + batchSize + " checkpointBatchTimeout=" + timeout + " partitions=" + + partitions); + } + + void run() throws Exception { + DefaultCamelContext context = DstCamel.newContext(dst); + DstExecutor downstream = dst.executor("downstream", Integer.MAX_VALUE); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("azure-eventhubs:dstns/dsthub?sharedAccessName=dst&sharedAccessKey=dst" + + "&checkpointBatchSize=" + batchSize + "&checkpointBatchTimeout=" + timeout) + .routeId(ROUTE) + .process(new AsyncProcessorSupport() { + @Override + public boolean process(Exchange exchange, AsyncCallback callback) { + // a non-blocking downstream call: answers after a latency chosen by the seed + String id = exchange.getMessage().getBody(String.class); + String partition = partitionOf(id); + long latency = dst.chance(0.1) + ? dst.between(timeout / 2, 2 * timeout) : dst.between(0, timeout / 2); + downstream.schedule("downstream answer " + id, latency, () -> { + if (dst.chance(0.05)) { + exchange.setException( + Dst.injected(new IllegalStateException("downstream failed for " + id))); + ledger.record("failed", partition, id); + } else { + // completed: the consumer's onComplete (processCommit) runs within done() + ledger.record("completed", partition, id); + } + callback.done(false); + afterExchange(partition); + }); + return false; + } + }); + } + }); + try (MockedStatic<EventHubsClientFactory> factory = Mockito.mockStatic(EventHubsClientFactory.class)) { + factory.when(() -> EventHubsClientFactory.createEventProcessorClient(any(), any(), any())) + .thenAnswer(inv -> { + processEvent = inv.getArgument(1); + return Mockito.mock(EventProcessorClient.class); + }); + context.start(); + consumer = (EventHubsConsumer) context.getRoute(ROUTE).getConsumer(); + try { + planArrivals(); + dst.runUntil(() -> arrivalsLeft == 0 && inflight == 0, "all events delivered and completed"); + // the partitions are quiet now: one batch timeout plus the slowest checkpoint store answer later, every + // completed event must have been checkpointed + dst.trace("quiet"); + dst.runFor(timeout + maxStoreLatency + 100); + dst.check(withProbes(Invariants.lastCompletedAcknowledgedWhenQuiet(ledger, "completed", "checkpoint"))); + } finally { + dst.dropPending(); + context.stop(); + } + } + } + + private Invariant withProbes(Invariant invariant) { + return Invariants.of(invariant.name(), () -> { + String v = invariant.violation(); + return v == null ? null : v + " | probes " + probes; + }); + } + + private static String partitionOf(String id) { + return id.substring(1, id.indexOf('-')); + } + + private void planArrivals() { + for (int p = 0; p < partitions; p++) { + String partition = String.valueOf(p); + DstExecutor sdk = dst.executor("eventhubs-sdk-p" + p, 1); + long t = dst.between(0, 50); + int n = 0; + int bursts = (int) dst.between(1, 4); + for (int b = 0; b < bursts; b++) { + int size = (int) dst.between(1, 2L * batchSize + 1); + for (int i = 0; i < size; i++) { + String id = "p" + partition + "-e" + n++; + arrivalsLeft++; + sdk.schedule("deliver " + id, t, () -> deliver(partition, id)); + t += dst.between(0, 20); + } + t += dst.between(0, 3 * timeout); + } + } + } + + private void deliver(String partition, String id) { + arrivalsLeft--; + inflight++; + PartitionContext pc = new PartitionContext("dstns.servicebus.windows.net", "dsthub", "$Default", partition); + EventContext ctx = new EventContext(pc, new EventData(id), new FakeCheckpointStore(partition, id), null); + ledger.record("delivered", partition, id); + processEvent.accept(ctx); + } + + /** Called by the downstream step's answer: the exchange is done (completed or failed). */ + private void afterExchange(String partition) { + inflight--; + probeTimerState(partition); + } + + private void probeTimerState(String partition) { + try { + ScheduledFuture<?> timer = this.<Map<String, ScheduledFuture<?>>> field("scheduledTasksByPartition").get(partition); + AtomicInteger count = this.<Map<String, AtomicInteger>> field("processedEventsByPartition").get(partition); Review Comment: Thanks, you were right, and it was worse than it looked. The probes wrapped the `NoSuchFieldException` in an `IllegalStateException`, but they run inside DST tasks and the checkpoint store, where the harness traces the throwable and `processCommit` logs and ignores it. With `processedEventsByPartition` renamed, all 200 seeds I ran still passed, with empty probes. In 0b1c0ebc3866 the world looks the two fields up when it is built, so a rename fails seed 1 with `EventHubsWorld reads the private field <name> of EventHubsConsumer, which no longer exists: update the world`. I kept the reflection rather than adding an accessor to `EventHubsConsumer` only for a test. _Claude Code on behalf of allthingssecurity_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
