[jira] [Created] (CAMEL-25655) camel-zookeeper - after the ZooKeeper session expired the consumer waits for ever without any Camel log line, and the producer fails every message, until the route is restarted

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

shashank created CAMEL-25655:
--------------------------------

             Summary: camel-zookeeper - after the ZooKeeper session expired the consumer waits for ever without any Camel log line, and the producer fails every message, until the route is restarted
                 Key: CAMEL-25655
                 URL: https://issues.apache.org/jira/browse/CAMEL-25655
             Project: Camel
          Issue Type: Bug
          Components: camel-zookeeper
            Reporter: shashank
            Assignee: shashank


A ZooKeeper session expires when the client cannot reach the ensemble for longer than the session timeout (the component's {{timeout}} option, 5000 ms by default): a network partition, a long GC pause, a suspended VM, an ensemble that loses its quorum for a few seconds. The server then expires the session; when the client reaches it again it is told so, the {{ZooKeeper}} handle is closed for good (state {{CLOSED}}), and every watcher of the handle gets an event of type {{None}} with state {{Expired}}. A new handle is needed.

Lines at main 1de5ec5135f8:
* {{FutureEventDrivenOperation.process}} (operations, 41-60 in the file) releases the waiting thread only for one of its awaited event types ({{NodeDataChanged}}/{{NodeDeleted}}, {{NodeCreated}}/{{NodeDeleted}}, {{NodeChildrenChanged}}); {{get()}} waits on the latch without a timeout. The {{Expired}} event is ignored.
* {{ConnectionHolder.getZooKeeper}} returns the cached handle whenever it is not null, also when it is closed; nothing in the module creates a new one while the route runs.

Consequences:
* Consumer: its operations thread waits in {{DataChangedOperation}} / {{ExistenceChangedOperation}} / {{ChildrenChangedOperation}} for an event that can no longer come. The ZooKeeper client logs one WARN ("Unable to reconnect to ZooKeeper service, session 0x... has expired"); after that the route stays started, consumes nothing, and Camel logs nothing (the {{Expired}} event is only logged at DEBUG). This hits the configuration for watching a node continuously: {{repeat=true}}.
* The same wait for ever when the expiry is noticed while no watch of the consumer is registered, for example while the operations thread routes an exchange: the next watch request on the closed handle is answered with {{SESSIONEXPIRED}} to the request's callback only ({{ClientCnxn.conLossPacket}}), the watch is not registered and no {{Expired}} event reaches it - and the callbacks of the three operations are empty. The longer the route takes, the more likely this is.
* Producer: every message fails with {{KeeperException.SessionExpiredException}} until the route is restarted.

A route stop / start creates a new consumer and the endpoint's connection manager a new handle ({{doStop}} closes it), so a route restart does recover; that is the only way out on main.

h3. Reproduction

New {{ZooKeeperConsumerSessionExpiryTest}} with a ZooKeeper server in the test JVM ({{ZooKeeperServer}} + {{NIOServerCnxnFactory}} from the zookeeper jar). The session of the Camel handle is expired the usual way: a second client opens the same session (id and password) and closes it. On main:
* consumer ({{repeat=true&backoff=200}}): the node is updated after the expiry; {{the consumer did not consume the update made after its session had expired ==> expected: <true> but was: <false> within 20 seconds}};
* the same, with the expiry while the route's processor holds the operations thread: {{the consumer did not consume the update made after its session had expired during an exchange ...}};
* consumer with {{repeat=false}}: {{the consumer did not read the data again after its session had expired ==> expected: <2> but was: <1> within 20 seconds}} (see the behaviour after the fix below);
* producer: a request after the expiry: {{the producer failed after its session had expired ==> expected: <null> but was: <org.apache.zookeeper.KeeperException$SessionExpiredException: KeeperErrorCode = Session expired for /camel-session-expiry>}}.
With the fix the module log shows the event reaching the operation ("The ZooKeeper session expired while waiting for an event on node ..."), the failed watch request ("Could not watch node ...") and the new session ("The ZooKeeper session 0x... can no longer be used (CLOSED). Creating a new session.").

h3. Proposed fix

* {{FutureEventDrivenOperation}}: on ({{None}}, {{Expired}}) stop waiting and fail with an {{ExecutionException}} caused by {{SessionExpiredException}}. The callbacks of the watch requests of the three operations fail the operation the same way when the result code is not {{OK}} ({{NONODE}} is the normal result of the exists watch and is not a failure there). The consumer hands the failure to its exception handler (one WARN per expiry) and starts the sequence of operations again after the configured {{backoff}}, as it does for any failed operation.
* {{ConnectionHolder.getZooKeeper}}: when the cached handle is no longer alive ({{CLOSED}} or {{AUTH_FAILED}}), close it and create a new handle; each handle counts down its own connection latch.
* {{ZooKeeperConsumer}}: take the connection again before the sequence is started again, and do not queue a copy of the failed operation in addition to the new sequence (on main every restart left that extra copy in the queue; after an expiry it would wait on the closed handle).
* {{ZooKeeperProducer}}: take the connection for each message (a cheap check when the handle is alive).

Behaviour after the fix: the WARN is "The ZooKeeper session expired while waiting for an event on node ...", "Could not watch node ..." when the expiry was noticed before the watch request, or "All operations exhausted without one result" (no cause) for a consumer of a node that does not exist yet ({{AnyOfOperations}} hides the cause). The restarted sequence reads the node again, so after every expiry the consumer emits the node's current data once more, also when it did not change and also with {{repeat=false}}; changes made during the outage are seen only as the latest value.

No option or default changes, no upgrade guide entry. The new test needs {{metrics-core}} and {{snappy-java}} with test scope (zookeeper declares them {{provided}}; the embedded server loads them); the {{snappy-java}} version is a new property in the parent pom.

camel-zookeeper unit tests with the fix: 10 run, 0 failures, 0 skipped. Not run: the module's integration tests (they need a container), an ensemble of several servers.

How it was found: a sweep of the reconnect code of the components, then a small TLA+ model (handle alive / closed, the operations thread waiting / in backoff, expiries, updates). For main it reaches the thread waiting on a closed handle after one expiry, and a run that never watches again; with the fix both properties hold for two expiries (27 states). The model is a sketch; the tests are the evidence.

Not changed: a stopped and started consumer instance (only reachable through a route policy or user code, not through a route restart) still adds no operations, because {{shuttingDown}} is never reset; the thread that creates a new handle waits for its first event without a timeout, as at start (a producer message sent while the ensemble is still unreachable after an expiry waits instead of failing at once).

Affected: main (1de5ec5135f8; tested). camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x have the same {{FutureEventDrivenOperation.process}}, {{ConnectionHolder.getZooKeeper}} and empty watch callbacks (read, not run).

Duplicate check (2026-10-14): JIRA text "zookeeper expired" (CAMEL-8208, route policy, 2015), component camel-zookeeper recent issues (cluster view and master only); GitHub pull requests "ZooKeeperConsumer", "zookeeper session expired", "camel-zookeeper consumer" (#25164 moved the backoff to BackgroundTask; nothing about expiry).

_Filed with Claude Code on behalf of allthingssecurity._




--
This message was sent by Atlassian Jira
(v8.20.10#820010)