Repository navigation
CAMEL-26957: camel-nats - a DST fake world for the JetStream pull consumer - #29995
Merged
davsclaus merged 2 commits intoNov 15, 2026
Merged
Conversation
davsclaus
reviewed
Nov 15, 2026
davsclaus
left a comment
davsclaus
left a comment
Contributor
There was a problem hiding this comment.
Thanks @allthingssecurity, there is a lot to like: the jnats proxies record any method they do not model as a violation, subscribe checks that the ack policy, AckWait and MaxDeliver reach the server, both paths for a deleted consumer are there (the 409 Consumer Deleted and the empty fetch followed by getConsumerInfo), the lost-ack excuse is written down, and there is a self-check against CAMEL-26884. Test-only, test-scoped deps, no Thread.sleep, and the catalog copies are identical.
One change I'd like, about how faithful the fake is:
- The fetch returns on the first message instead of holding a partial batch. The [P] Javadoc (FakeJetStream.java:71-72) says the fetch "waits up to the timeout for up to batch messages and returns what it got", which is what jnats does, but the wait at FakeJetStream.java:178 ends as soon as one message is available. In jnats 2.26.4,
NatsJetStreamPullSubscription._fetchsends one pull withexpiresInand loops whilebatchLeft > 0 && timeLeft > 0, so a partial batch comes back only at expiry, when the batch is full, or on a terminus status. The server sends each message as soon as it is available, so its AckWait already runs during that hold, while the fake starts AckWait only when the fetch returns (FakeJetStream.java:198). With the defaultpullBatchSize=10this is the common path, and seeds withpullFetchTimeout=5000andackWait=5000are where AckWait runs out before the route sees the message on a real server. A sketch:with// [P] jnats holds a partial batch until the pull expires; the server sends each message when it is // available, so its AckWait runs from then long fetchStart = dst.now(); dst.awaitUntil(() -> !active || deleted || availableCount(startSeq) >= batch, Math.max(1, timeout), name + " fetch"); ... m.redeliverAt = Math.max(fetchStart, m.availableSince) + ackWaitMillis;
availableSinceset at publish, release and the end of a nak delay. After that the seeds and the CAMEL-26884 self-check need a re-run, and any new failing seed triaged (fake or consumer).
Smaller points, optional:
- No message published once the faults cleared. PubsubWorld (#29996) publishes an END message then and requires it to be acked, which catches a pull loop that ended. Here, if the fetch loop exits after the last message was acked (e.g. the
IllegalStateExceptionreturn at NatsConsumer.java:494-497), the world still passes. One morenats.publish(...)in the-2step of the finisher (NatsWorld.java:226-232), required to end ACKED, would close that gap. - The injected fetch failure is an
IllegalArgumentException(FakeJetStream.java:173-177), which jnats does not throw for a valid fetch. During a disconnect the pull is buffered and the fetch ends empty at its timeout; a 409 thatPullMessageManager.manageStatusreports as an error comes out as aJetStreamStatusException, which would also reach theelsebranch at NatsConsumer.java:484-490 that no fault drives today. - The server restart. In jnats 2.26.4 a 409 "Server Shutdown" is handled as a warning and the fetch keeps waiting, so on a real restart a waiting fetch ends empty at its timeout; 409 Consumer Deleted comes when the consumer is deleted while connected. If the seed picks between the two at FakeJetStream.java:183-185, the
getConsumerInfopath is also reached from a waiting fetch. - Follow-up idea, not for this PR: #29998 (CAMEL-26915) fixes a stop and start of the same consumer while its task is still subscribing. This world only stops the route, which makes a new consumer each time, so it cannot reach that bug. Once #29998 is merged, an operator step that stops and starts the route's consumer (
ServiceHelper.stopService+startService, sometimes right after a start) would let the world fail on the code before #29998. - Nit: the last invariant in
@DstWorld(NatsWorld.java:80-81) ends with "(or skipped by a consumer made again after its ack was lost)", but thedst.checkname (NatsWorld.java:174) does not. The guide wants the same names.
Note: GitHub has not started any CI run on this PR yet; it needs one before merge.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
davsclaus
force-pushed
the
camel-nats-dst-world-pr
branch
from
November 15, 2026 07:25
828c0df to
d57b040
Compare
Contributor
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
Contributor
|
🧪 CI tested the following changed modules:
🔁 1 test passed only after a retry on JDK 21 (1 retried attempt) Recovered flaky tests on JDK 21 (1)
🔬 Scalpel shadow comparison — Scalpel: 20 of 726 tested, 3 compile-only — current: 20 all testedMaveniverse Scalpel detected 20 affected modules (current approach: 20). Skip-tests mode would test 20 modules (3 direct + 0 downstream), skip tests for 3 (generated code, meta-modules) Modules Scalpel would test (20)
Modules with tests skipped (3)
All tested modules (47 modules, 9m 10s total)Total reactor time: 9m 10s
Top 20 slowest modules:
|
Contributor
Author
|
Thanks @davsclaus. Pushed 80a2391 on top of your rebase:
The faults list of Claude Code on behalf of allthingssecurity |
gnodet-bot
approved these changes
Nov 15, 2026
gnodet-bot
left a comment
gnodet-bot
left a comment
Contributor
There was a problem hiding this comment.
Well-constructed DST world for the NATS JetStream pull consumer. The fake models real NATS semantics faithfully.
Items from davsclaus's review — all confirmed addressed:
- fetch() awaitUntil — All three
awaitUntilcalls include!activeas a termination condition; the full-batch wait at line 271 uses!active || deleted || availableCount(startSeq) >= batch, correctly holding a partial batch until expiry or completion, withredeliverAtanchored toavailableSince. - ack() state handling —
ack()accepts PENDING and AVAILABLE (lines 482-483); double-ack (already ACKED) is silently ignored, consistent with the comment "an ack of an older delivery is accepted." Intentional design, documented. - nak() stale-delivery guard —
nak()guards withm.state == State.PENDING && m.deliveries == number(lines 490-492), preventing stale naks from resetting the current delivery. The asymmetry withack()is correct per NATS semantics. - subscribe() PENDING reset race —
subscribe()resets PENDING→AVAILABLE (lines 589-590);stopWhenIdle()checksinRoute > 0and reschedules if an exchange is in-flight (lines 960-966), preventing premature reset. - restartServer() 409 paths — Both paths confirmed:
deleted && serverShutdown→ warning + continue waiting;deletedalone →JetStreamStatusException(409, CONSUMER_DELETED)(lines 276-290). - settled() ack-lost + new consumer — Returns true for ACKED/GIVEN_UP, or ack-lost where
allMatch(s -> s.startSeq > m.seq) && anyMatch(s -> s.active && !s.deleted)(lines 416-421); theanyMatchguard prevents vacuous truth on an empty stream.
Test-only PR, test-scoped dependencies, no Thread.sleep, pom.xml vmargs use the standard byte-buddy-agent pattern.
Claude Code on behalf of gnodet. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
davsclaus
approved these changes
Nov 15, 2026
davsclaus
left a comment
davsclaus
left a comment
Contributor
There was a problem hiding this comment.
Thanks @allthingssecurity, 80a2391 covers every point. I read it against my review:
- The fetch now holds a partial batch:
awaitUntil(... availableCount(startSeq) >= batch, wait, ...), andtakesetsredeliverAt = Math.max(fetchStart, m.availableSince) + ackWaitMillis.availableSinceis set at publish, ingiveUpOrRelease(the end of AckWait, and of a nak delay, sincenaksetsredeliverAtbefore it releases), and when a new ephemeral consumer makes a message available again. So AckWait can run out while the fetch holds a message, as on a real server. - The
endmessage is published in the-2step, and the finisher and the invariant require it to end ACKED, so a fetch loop that ended now fails the seed. - The fetch fault is a fetch that ends empty at its timeout while the connection is down. Keeping the
elsebranch of the catch undriven, rather than inventing a 409 that 2.26.4 does not report as an error, is the right call. restartServerpicks Server Shutdown (a warning: it keeps waiting to the timeout and returns what it took) or Consumer Deleted (it throws), and the 503 on the next fetch reachesgetConsumerInfofrom both.- Noted as a follow-up after #29998 (CAMEL-26915: stop and start of the same consumer).
- The invariant names now match.
The re-run (5000 of 5000 seeds pass on main, 339 of 2000 fail on the CAMEL-26884 self-check) is the evidence I hoped for. Looks good to me.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
…sumer Adds a camel-test-dst world for the camel-nats JetStream pull consumer, in the dst package of its tests: NatsConsumerDstTest (1000 seeds), NatsWorld and FakeJetStream. The real consumer runs against a fake NATS server with one JetStream stream, given as a shared jnats Connection with the connection option, written from the NATS documentation and the jnats client: pull fetches that wait up to their timeout for up to a batch, explicit ack, nak and nak with a delay, AckWait redelivery, MaxDeliver, ephemeral and durable consumers, a server restart that deletes the ephemeral consumers (409 Consumer Deleted for a waiting pull, an empty fetch and no consumer info afterwards), a flush that times out, and acks that are lost. The seed picks pullBatchSize, pullFetchTimeout, ackWait, maxDeliver, nackWait, a durable or an ephemeral consumer and the fault rates, and injects failed fetches, lost acks, flushes that time out, failed and slow processing (now and then longer than AckWait), route restarts (also while the consumer waits in a fetch) and server restarts. Invariants: the consumer passes its ack policy, AckWait and MaxDeliver to the server and uses the connection as documented; a message is acked only after its exchange completed and nak'd only after it failed; a stopped consumer leaves no subscription active; once the faults cleared every message is acked, or given up after MaxDeliver (or skipped by a consumer made again after its ack was lost). Not explored: push subscriptions and core NATS subjects, a stop between two messages of a fetched batch, the interrupt of the consumer thread by the stop, and a start that comes while the stopped consumer task still subscribes. Self-check: on the code before CAMEL-26884 (6bc46f9 reverted) 346 of 2000 seeds fail; main passes 5000 of 5000 seeds. The module's pom loads the byte code agent of camel-test-dst, the test log is capped, and the catalog lists the world. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…h, AckWait from availability Review of davsclaus: - As jnats 2.26.4 NatsJetStreamPullSubscription._fetch: the fake fetch holds a partial batch until the pull expires, the batch is full or a status ends it, and the AckWait of a message runs from when it became available (published, released, end of a nak delay), or from the start of the fetch if later, as the server sends it then. - A message published once the faults cleared must end acknowledged, so a fetch loop that ended is caught. - The injected fetch fault is a fetch that ends empty at its timeout while the connection is down, instead of an IllegalArgumentException. - The seed picks whether a waiting fetch sees 409 Server Shutdown (a warning: it keeps waiting and returns what it got at its timeout) or 409 Consumer Deleted when the ephemeral consumers are deleted. - The last invariant name of dst.check matches @DstWorld. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
davsclaus
force-pushed
the
camel-nats-dst-world-pr
branch
from
November 15, 2026 13:32
80a2391 to
527beec
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
CAMEL-26957, part of CAMEL-26610. Two differences from the JIRA text: the world drives the pull consumer only, as the push consumer and core NATS subjects deliver on threads of jnats that the harness does not run; and a server restart is modelled as the server deleting the ephemeral consumers while the connection stays up, not as a reconnect of the connection itself.
A deterministic simulation test (camel-test-dst) of the camel-nats JetStream pull consumer, in the
dstpackage of its tests, in the same form as the SQS world:NatsConsumerDstTest(1000 seeds),NatsWorldandFakeJetStream. Test only; no main code changes.What the world models: the real consumer (
NatsConsumerand its consuming task) and Camel's routing run on the virtual-time loop; only the NATS server is faked, behind a jnatsConnectiongiven to the endpoint with theconnectionoption (dynamic proxies of the jnats interfaces, Mockito mocks of the jnats value classesServerInfo,StreamInfo,ConsumerInfoandNatsJetStreamMetaData). The fake is written from the NATS JetStream documentation and the jnats client, and cites them: a pullfetchholds a partial batch until its pull expires, the batch is full or a status ends it, as jnats 2.26.4NatsJetStreamPullSubscription._fetchdoes, and the AckWait of a message runs from when the server could send it (when it became available, or the start of the fetch), explicitack,nakandnakWithDelay, redelivery after AckWait, MaxDeliver, ephemeral and durable consumers, the deletion of the ephemeral consumers, where the seed picks between a server restart (a waiting pull gets 409 Server Shutdown, a warning, and returns at its timeout what it got) and a deletion while connected (a waiting pull gets 409 Consumer Deleted); a later pull gets no message andgetConsumerInfoanswers null, so the consumer finds the consumer gone and subscribes again), aflushthat times out, and acks that are lost (an ack is a publish without an answer). A new ephemeral consumer starts with no delivery state of its own; the fake keeps one state per message for all consumers, which is written down where it matters.The seed picks
pullBatchSize,pullFetchTimeout,ackWait,maxDeliver,nackWait, a durable or an ephemeral consumer and the fault rates, and injects fetches that end empty at their timeout while the connection is down, lost acks, flushes that time out during the stop, failed and slow processing (now and then longer than AckWait), route restarts (also while the consumer waits in a fetch) and server restarts.Invariants: the consumer passes its ack policy, AckWait and MaxDeliver to the server and uses the connection as documented (it never closes the shared connection); a message is acked only after its exchange completed, and nak'd only after it failed; a stopped consumer leaves no subscription active, checked when the route starts again and after the final stop; once the faults cleared every message is acked, or given up after MaxDeliver, and a message published then is acked (so a fetch loop that ended is caught). One case is accepted: when an ack was lost and the server then deleted the ephemeral consumer, the consumer subscribes again after that message, as it counts it as acknowledged.
Limits, written in the world's Javadoc: the operator stops the route only while no exchange is in it, so a stop between two messages of a fetched batch is not driven; the harness interrupts no thread, so the interrupt part of CAMEL-26884 is not checked (its flush and subscribe parts are); the route starts again 100 ms or more after a stop, so a start that comes while the stopped consumer task still subscribes is not driven. Not explored: push subscriptions and core NATS subjects (their dispatchers deliver on jnats threads), manualAck, AckAll and AckNone, a connection created by the consumer itself.
Other changes:
camel-test-dstandmockito-coreas test dependencies of camel-nats, and the byte code agent of camel-test-dst on the surefire command line, as for the other worlds.@DstWorldannotation lists the world in the component JSON (and its catalog copy) and in the table of the camel-test-dst page (and its catalog copy, byte-identical).Tests:
camel-nats-consumer-run-token) also changesNatsConsumer: this branch merges with it without a conflict, and no invariant of the world depends on it. Merged locally with that branch, the world passes 2000 of 2000 seeds.What was not run locally: the ITs (they need Docker) and the full root build. I built and tested offline with the versions cached on my machine: commons-lang3 3.21.0 and Guava 33.7.2 (main has 3.22.0 and 33.8.0-jre).
Target
mainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.(I built and tested camel-test-dst and camel-nats, including the formatter and import-sort plugins. The generated files touched are the component JSON with its catalog copy, and the table of the camel-test-dst page with its catalog copy (byte-identical). I did not run the full root build.)
AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a
Co-Authored-Bytrailer.Claude Code on behalf of allthingssecurity
🤖 Generated with Claude Code