allthingssecurity opened a new pull request, #27858: URL: https://github.com/apache/camel/pull/27858
# Description [CAMEL-25655](https://issues.apache.org/jira/browse/CAMEL-25655) When the ZooKeeper server expires the session of a client (the client could not reach the ensemble for longer than the session timeout, the `timeout` option, 5 seconds by default), the `ZooKeeper` handle is closed for good and every watcher gets an event of type `None` with state `Expired`. A new handle is needed. What went wrong on main: 1. Consumer: `FutureEventDrivenOperation.process` releases the waiting thread only for its awaited event types, so the `Expired` event was ignored and the operations thread waited for ever for a change that could no longer come. After the one WARN of the ZooKeeper client (`Unable to reconnect to ZooKeeper service, session 0x... has expired`) the route stayed started and consumed nothing; Camel logged nothing (the event is logged at DEBUG only). 2. 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 longer the route takes, the more likely). The next operation then sends its watch request on the closed handle; ZooKeeper answers it with `SESSIONEXPIRED` (`CONNECTIONLOSS` when the handle lost its connection, `AUTHFAILED` after an authentication failure) to the request's callback only, does not register the watch, and sends no `Expired` event to it. The callbacks in `DataChangedOperation`, `ExistenceChangedOperation` and `ChildrenChangedOperation` were empty. 3. `ConnectionHolder.getZooKeeper` returned the cached handle also when it was closed, so the producer failed every message with `SessionExpiredException` until the route was restarted. A route restart does recover (the new consumer gets a new handle); nothing else did. Changes: - `FutureEventDrivenOperation`: on (`None`, `Expired`) stop waiting and fail with an `ExecutionException` caused by `SessionExpiredException`. New protected `watchRequestCompleted(int rc)`, called by the callbacks of the watch requests of the three operations: for any result code other than `OK` the operation stops waiting and fails with the `KeeperException` of that code. `ExistenceChangedOperation` does not pass `NONODE` on: a node that does not exist is what it waits for, and ZooKeeper registers the exists watch then too. `DataChangedOperation` and `ChildrenChangedOperation` now also fail on `NONODE` (the node was deleted between the read and the watch request): on main no watch was registered then and the thread waited for ever; now the sequence starts again and waits for the node to be created. - 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 already 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 one. Each handle counts down its own connection latch, so a late event of the old handle cannot release the wait for the new one, and the thread that created a handle waits on that handle's latch. Creation is synchronized, the wait for the connection is not, so a stop can still close a handle that is connecting. - `ZooKeeperConsumer`: take the connection again before the sequence of operations is started again, and do not queue a copy of the failed operation in addition to the new sequence. On main every restart of the sequence left that extra copy in the queue; after an expiry the copy would wait on the closed handle. - `ZooKeeperProducer`: take the connection for each message (one `getState()` check while the handle is alive). - Test scope only: `metrics-core` (`${metrics-version}`) and `snappy-java`, which the zookeeper artifact declares `provided` and which the ZooKeeper server of the new test loads. `snappy-java` had no version in the Camel parent; it gets a new `snappy-java-version` property in `parent/pom.xml`, like other test-only libraries (for example `wiremock-version`). What users can notice: - After a session expiry the consumer logs one WARN and an INFO about the new session, and consumes again after `backoff`; the producer sends again. The WARN is `The ZooKeeper session expired while waiting for an event on node ...` when the consumer was waiting for a change, `Could not watch node ...` when the expiry was noticed before the watch request, and `All operations exhausted without one result` (no cause) when it was waiting for 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; intermediate values are lost, as with any one-shot ZooKeeper watch. - A watch request that the client fails with `CONNECTIONLOSS` (the connection was lost before the request was answered, the session not yet expired) now also fails the operation, and the sequence is started again after `backoff`, one WARN per attempt; on main no watch was registered for it and the thread waited for ever. This path was read in the ZooKeeper client, not tested. - No option, default or output format changes, so there is no upgrade guide entry. Tests: - New `ZooKeeperConsumerSessionExpiryTest` (4 tests) with a ZooKeeper server in the test JVM (`ZooKeeperServer` and `NIOServerCnxnFactory` from the zookeeper artifact). The session of the Camel handle is expired the usual way: a second client opens the same session and closes it. Each test starts only its own route; every wait is bounded (20 seconds), no sleeps. - Consumer with `repeat=true&backoff=200`: the node is updated after the expiry, and the update must be consumed. Then, once the operations thread waits for a change, the node is updated again through the consumer's own session (ZooKeeper handles one session's requests in order, so the watch is registered first), and that update must be consumed too: it can only come from the watch on the new session. - The same, but the session expires while the route's processor holds the operations thread, so no watch is registered at the time of the expiry and the next watch request fails on the closed handle. - Consumer with `repeat=false`: after the expiry the unchanged data is emitted once more. - Producer: a request after the expiry must succeed and write the node. - Without the change (main's sources of camel-zookeeper, same test class) all four fail: `the consumer did not consume the update made after its session had expired ==> expected: <true> but was: <false> within 20 seconds`, `... after its session had expired during an exchange ...`, `the consumer did not read the data again after its session had expired ==> expected: <2> but was: <1> within 20 seconds` and `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 everything but the callback change, only the test with the blocked processor fails (`the consumer did not consume the update made after its session had expired during an exchange`). - With the change, the camel-zookeeper unit tests pass: 10 run, 0 failures, 0 skipped (the session-expiry class also passed three more times in a row). What is not changed: - A consumer instance that is stopped and started again (a route policy or user code; a route restart creates a new consumer) still adds no operations after the start, because `shuttingDown` is never reset. - The producer and the consumer of the same endpoint URI share one handle, and stopping one closes it for the other, as before. The producer now gets a new handle on its next message; a consumer waiting on a watch of that handle is not woken by a close (only by an expiry), as before. - The thread that creates a new handle waits for its first event without a timeout, as at start on main. For the producer this means a message sent while the ensemble is unreachable after an expiry now waits until the client connects, where on main it failed at once with `SessionExpiredException`. - A stop between the consumer's `isRunAllowed()` check after the backoff and `getConnection()` can create a handle after `doStop` closed the holder; that handle is not closed (a very small window, not handled). What was not run: the module's integration tests (they need a container), an ensemble of several servers, an expiry caused by a real network partition. Notes: - How it was found: a sweep of the reconnect code of the components, then a small TLA+ model (handle alive or closed, the operations thread waiting or in backoff, expiries, updates). For main it reaches the thread waiting on a closed handle after one expiry; with the change the consumer watches again on a live handle in the end. The model does not cover a watch request made after the expiry (the callback change). The model is only a sketch; the tests are the evidence. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). (The ticket is drafted and not filed yet; CAMEL-25655 is a placeholder until it is.) # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built and tested camel-zookeeper, including the formatter and import-sort plugins. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) 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-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
