allthingssecurity opened a new pull request, #27762: URL: https://github.com/apache/camel/pull/27762
# Description [CAMEL-25582](https://issues.apache.org/jira/browse/CAMEL-25582) When the WebSocket of a consumer with `consumeAsClient=true` is closed, the close handler starts a periodic timer, and every run of the timer starts a connect attempt on a worker thread (`vertx.setPeriodic(..., timerId -> vertx.executeBlocking(() -> { getEndpoint().getWebSocket(...); vertx.cancelTimer(timerId); ... }, false))`). `VertxWebsocketEndpoint.getWebSocket` waits for the handshake for up to the connect timeout of the client options (60 seconds by default). The timer runs again after `reconnectInterval` (1 second by default) and nothing checks whether the previous attempt has ended, so the attempts run in parallel, up to the size of the Vert.x worker pool (20 by default). 1. **Duplicate WebSockets and duplicate delivery.** When the server answers the handshake later than `reconnectInterval` (a TCP connect with a retransmitted SYN, a proxy that waits for its backend, a server that is still starting), all the attempts in flight succeed. Each registers the handlers of the consumer on its own WebSocket. The endpoint keeps the last one; the others stay open until the endpoint stops, and every message the server sends to its clients is then routed once per WebSocket. When these WebSockets are closed later, each close handler starts its own timer. 2. **Reconnecting does not stop with the consumer.** The consumer has no `doStop`; the timer is only cancelled by a successful attempt or by `maxReconnectAttempts` (unlimited by default). A route that is stopped while the server is away goes on connecting every interval, and the WebSockets of the attempts that succeed after the stop stay open (the test shows the open WebSockets; whether messages on them are still routed was not tested). 3. A connect that completes after the caller stopped waiting (connect timeout) still registers the handlers, and nobody keeps or closes its WebSocket. This change: - `VertxWebsocketClientConsumer` has one reconnect timer: a close event while the timer runs does not start another one, and a WebSocket that is found closed when its attempt returns starts the timer again; - a run of the timer is skipped while an attempt is in flight; - `doStop` cancels the timer, and a run of the timer of a stopped consumer cancels it as well; - start and reconnect connect under one lock. The lock belongs to the endpoint, not to the consumer: a route that is stopped and started again gets a new consumer (`DefaultRoute` calls `endpoint.createConsumer` again), while an attempt of the previous consumer can still wait for its handshake. A start therefore waits for that attempt (at most the connect timeout) instead of connecting next to it; - an attempt that returns after its consumer was stopped closes the WebSocket it opened. It does this before it releases the lock, so a start that waits for the lock finds the WebSocket closed and connects again, and cannot take over a WebSocket that is about to be closed. An attempt that only found a WebSocket that was already open does not close it; - `VertxWebsocketEndpoint.getWebSocket` closes a WebSocket whose connect completes after the caller stopped waiting, instead of registering the handlers on it; the same when the setup callback throws. The `webSocket` field is now volatile. Why `doStop` of the consumer does not close a WebSocket that is open and healthy: the WebSocket belongs to the endpoint, not to the consumer. It is the `webSocket` field of `VertxWebsocketEndpoint` (line 72), `VertxWebsocketEndpoint.doStop` closes it (lines 119-123), and a producer on the same endpoint sends on the same WebSocket: `VertxWebsocketProducer` line 141 calls `endpoint.getWebSocket(exchange)` (lines 231-232), which returns that field when it is open (line 191). Closing it in the consumer would cut the connection of such a producer. So the consumer only closes what a reconnect attempt of its own opened after the stop. Left as on main, possible follow-ups: 1. Stopping the route does not close a WebSocket that is open; it is closed when the endpoint stops. 2. The producer side of `getWebSocket` is not serialised: concurrent exchanges that find the WebSocket closed can each open one, and the endpoint keeps the last. Behaviour change, with an entry in the 4.24 upgrade guide (`camel-4x-upgrade-guide-4_24.adoc`; if that file is not on main yet, this PR creates it with the header of the 4_23 guide and adds it to the list in `camel-4x-upgrade-guide.adoc`): one attempt at a time, so with a server that does not answer the attempts follow each other (each up to the connect timeout) instead of running in parallel, and `maxReconnectAttempts` counts them one after the other; a stopped consumer no longer reconnects. Not changed, possible follow-ups: stopping the route does not close a WebSocket that is open (as on main; it is closed when the endpoint stops), and the producer side of `getWebSocket` is not serialised, so concurrent exchanges that find the WebSocket closed can each open one. Tests: - New `VertxWebsocketConsumerAsClientSlowReconnectTest` (4 tests). The server is a plain Vert.x HTTP server in the test that counts handshakes and WebSockets and can accept a handshake at once, accept it after 1 second, hold it until the test answers it, or reject it. The consumer uses `reconnectInterval=50`. - Without the change (the test class with the two main classes of `main`) all 4 fail: - one lost connection, the server answers after 1 second: `handshakes sent for one reconnect ==> expected: <1> but was: <20>`, `WebSockets the consumer has open after one reconnect ==> expected: <1> but was: <20>` and `mock://result Received message count. Expected: <1> but was: <20>` (one message sent by the server on each WebSocket it has); - route stopped while a reconnect is in flight: `WebSockets pending or open after the consumer was stopped ==> expected: <0> but was: <20>`; - the server rejects the handshake, the route is stopped: `Condition with alias 'no reconnect attempt after the consumer was stopped' didn't complete within 10 seconds` (the handshakes go on); - route stopped and started again while the handshake of a reconnect is held, then the server answers: `handshakes sent by the reconnect and the restart ==> expected: <2> but was: <6>`, `WebSockets the consumer has open after the restart ==> expected: <1> but was: <6>` (6 in this run; the number depends on how many timer runs fit in before the stop). - With the change the restart test sees two handshakes (the pending attempt, whose WebSocket is closed, and the start), one open WebSocket, and one delivery of a message sent afterwards. - With the change, camel-vertx-websocket passes: 107 tests, 0 failures. How it was found: a small TLA+ model of the reconnect on main (close events, timer runs, attempts in flight, stop), with the assumption that an attempt can still be in flight when the timer runs again. The model checker finds traces of the model in which the consumer is registered on two open WebSockets (close, two timer runs, both attempts succeed) and in which the timer runs after the stop. That only reproduces the defect on a model of main; it is not a proof of the change, and the model has no restart. The evidence for the change is the tests above, run against the real consumer. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] 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). # 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 the affected module, 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]
