shashank created CAMEL-25815:
--------------------------------
Summary: camel-nats - the JetStream pull consumer stops consuming
for good when the server deleted its ephemeral consumer (after a disconnect of
more than 5 s or a server restart)
Key: CAMEL-25815
URL: https://issues.apache.org/jira/browse/CAMEL-25815
Project: Camel
Issue Type: Bug
Components: camel-nats
Reporter: shashank
Assignee: shashank
With {{jetstreamEnabled=true}} the consumer uses a pull subscription by default
({{pullSubscription=true}}), and without {{durableName}} it subscribes with an
ephemeral JetStream consumer ({{NatsConsumer.setupJetStreamConsumer}}, main
62c9a3ebe409, lines 202-238). The NATS server deletes an ephemeral consumer
that had no waiting pull request for its inactive threshold (nats-server
{{consumer.go}}: {{JsDeleteWaitTimeDefault = 5 * time.Second}} plus up to one
second of jitter, once no ack is pending or the latest ack deadline plus the
threshold passed). That happens when the client is disconnected for more than
about 5 seconds and reconnects (jnats reconnects for about 2 minutes by
default), and a server restart loses an ephemeral consumer as well.
jnats does not give a pull subscription a new consumer. After the deletion a
fetch ends with an empty list: the pull request has no responder (status 503,
which {{PullMessageManager.manageStatus}} of jnats 2.26.4 treats as the quiet
end of the pull). A pull request that was waiting at the deletion gets {{409
Consumer Deleted}}, which {{_fetch}} throws as {{JetStreamStatusException}};
{{runPullFetchLoop}} (lines 265-296) reports it to the exception handler and
continues, with the comment "so the consumer can recover when the consumer
state is restored", but nothing restores an ephemeral consumer. The route stays
started, nothing is consumed again and, after the one 409 at most, nothing is
logged. The messages stay in the stream until the route is restarted (which
then creates a new ephemeral consumer from its deliver policy, by default the
whole stream).
*Fix*: the pull loop asks the server for its consumer ({{getConsumerInfo}})
when a fetch returned nothing, at most every 5 seconds, and handles {{409
Consumer Deleted}}. When the consumer is gone (404) it subscribes again. An
ephemeral consumer is created again starting at the first stream sequence the
route has not acknowledged (or after the last one fetched), so that nothing is
skipped and the stream is not replayed from the deliver policy; a durable
consumer is created again from its configuration. A failed check (not
connected, timeout) changes nothing. No option changes.
*Tests*: new {{NatsConsumerPullConsumerDeletedTest}} (2 tests) with a fake
JetStream server behind the jnats interfaces the consumer calls
({{Connection}}, {{JetStream}}, {{JetStreamSubscription}}, {{Message}}):
{{consumesAgainAfterTheServerDeletedTheEphemeralConsumer}} (fetches then get
nothing) fails on main with {{mock://result Received message count. Expected:
<2> but was: <0>}}; {{consumesAgainAfterAWaitingPullGotConsumerDeleted}} (one
409, then nothing) fails on main with {{Expected: <1> but was: <0>}}. Both pass
with the fix (the first also checks that the message acknowledged before the
deletion is not routed again). camel-nats unit tests with the fix: 7 run, 0
failures.
*How it was found*: deterministic simulation testing with a fake JetStream
world (camel-test-dst, CAMEL-25738): the real consumer against a fake server
with connection outages (0.5 to 20 s) and restarts, ephemeral and durable
consumers, a slow and sometimes failing route. 274 of 1000 seeds failed on main
(some published messages never routed), first failing seed 2; all failing seeds
use an ephemeral consumer and all of them had the consumer deleted after an
outage or a restart. With the fix 0 of 1000. Then the JUnit test above. The
server and jnats behaviour were taken from the nats-server source and the jnats
2.26.4 byte code; not run against a real NATS server.
*Affected*: main, camel-4.23.x, camel-4.22.x and camel-4.18.x (the same pull
loop; read, not run).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
