[jira] [Created] (CAMEL-25815) 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)

["shashank (Jira)" <ji...@apache.org>]

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)