[
https://issues.apache.org/jira/browse/CAMEL-25582?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25582:
--------------------------------
Fix Version/s: 4.22.2
> camel-vertx-websocket - a consumeAsClient consumer starts overlapping
> reconnect attempts: after a slow handshake it holds up to 20 WebSockets and
> routes every message once per WebSocket; it also keeps reconnecting after its
> route is stopped
> ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25582
> URL: https://issues.apache.org/jira/browse/CAMEL-25582
> Project: Camel
> Issue Type: Bug
> Components: camel-vertx-websocket
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.22.2, 4.24.0
>
>
> When the WebSocket of a consumer with {{consumeAsClient=true}} is closed, its
> close handler starts a periodic timer, and every run of the timer starts a
> connect attempt on a worker thread
> ({{VertxWebsocketClientConsumer.configureWebSocketHandlers}}, lines 52-85 at
> main c3829069cb72):
> {code:java}
> vertx.setPeriodic(configuration.getReconnectInitialDelay(),
> configuration.getReconnectInterval(), timerId -> {
> vertx.executeBlocking(() -> {
> getEndpoint().getWebSocket(this::configureWebSocketHandlers);
> vertx.cancelTimer(timerId);
> return null;
> }, false)
> {code}
> {{VertxWebsocketEndpoint.getWebSocket}} (lines 164-203) connects and waits
> for the handshake for up to the connect timeout of the client options (60
> seconds by default, line 200). The timer runs again after
> {{reconnectInterval}} (1 second by default), and nothing checks whether the
> previous attempt has ended: {{executeBlocking(..., false)}} runs the attempts
> 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 or ingress that waits for its backend, a server
> that is still starting), all the attempts in flight succeed. Each of them
> registers the handlers of the consumer on its own WebSocket (the setup
> callback, line 188). The endpoint keeps the last one in its field; the others
> stay open until the endpoint stops. From then on every message the server
> sends to its clients is routed once per WebSocket. When these WebSockets are
> closed later, each close handler starts a timer of its own, so the next
> reconnect starts with that many timers.
> *2. Reconnecting does not stop with the consumer.* The consumer has no
> {{doStop}}. The timer is only cancelled by a successful attempt (line 62) or
> by {{maxReconnectAttempts}} (0 = 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 (messages on
> them were not tested).
> *3.* A connect that completes after the caller stopped waiting (the
> {{future.get}} timed out, line 200) still registers the handlers (line 188)
> and its WebSocket is not kept or closed by anyone.
> h3. Reproduction
> 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 route is
> {{from("vertx-websocket:localhost:<port>/feed?consumeAsClient=true&reconnectInterval=50")}}.
> On main all 4 fail:
> * {{reconnectSlowerThanTheIntervalOpensOneWebSocket}}: the server closes the
> WebSocket and answers the next handshakes after 1 second; then it sends one
> message on every WebSocket it has. {{handshakes sent for one reconnect ==>
> expected: <1> but was: <20>}}, {{WebSockets the consumer has open after one
> reconnect ==> expected: <1> but was: <20>}}, {{mock://result Received message
> count. Expected: <1> but was: <20>}}.
> * {{reconnectThatCompletesAfterStopIsClosed}}: the route is stopped while a
> reconnect is in flight. {{WebSockets pending or open after the consumer was
> stopped ==> expected: <0> but was: <20>}}.
> * {{stoppedConsumerDoesNotReconnect}}: 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).
> * {{restartWhileReconnectIsPendingOpensOneWebSocket}}: the route is stopped
> and started again while the server holds the handshake of a reconnect, 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).
> h3. Proposed fix
> In {{VertxWebsocketClientConsumer}}:
> * one reconnect timer per consumer: a close event while the timer runs does
> not start another one (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 too;
> * start and reconnect connect under one lock of the endpoint (a route that is
> stopped and started again gets a new consumer, while an attempt of the
> previous one can still wait for its handshake);
> * an attempt that returns after its consumer was stopped closes the WebSocket
> it opened, before it releases that lock, so that a start which waits for it
> connects again.
> In {{VertxWebsocketEndpoint.getWebSocket}}: a connect that completes after
> the caller stopped waiting closes its WebSocket instead of registering the
> handlers; the same when the setup callback throws.
> Behaviour change, with an entry in the 4.24 upgrade guide: one attempt at a
> time, so with a server that does not answer, 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 (follow-ups): stopping the route does not close a WebSocket that
> is open (as on main, it is closed when the endpoint stops; it belongs to the
> endpoint and a producer on the same endpoint uses it too); the producer side
> of {{getWebSocket}} is not serialised, so concurrent exchanges that find the
> WebSocket closed can each open one.
> camel-vertx-websocket with the fix: 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. This reproduces the defect on a model of main; it
> is not a proof of the fix. The defect was then confirmed with the real
> consumer as described above.
> Affected: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x.
> Verified by reading {{VertxWebsocketClientConsumer}} on each branch: the same
> {{setPeriodic}} / {{executeBlocking}} / connect / {{cancelTimer}} at lines
> 59-62, and no {{doStop}}. Not run on the maintenance branches.
> Duplicate check (2026-10-12): JIRA text "vertx-websocket reconnect" (1:
> CAMEL-23889, flaky tests), "VertxWebsocketClientConsumer" (1: CAMEL-20011,
> deprecated executeBlocking), "consumeAsClient" (1: CAMEL-20656, query
> parameters), "vertx-websocket reconnectInterval" (none), "vertx-websocket
> consumeAsClient duplicate" (none); GitHub pull requests "vertx-websocket
> reconnect": #24393 (merged, CAMEL-23889), #24740 (merged, CAMEL-24126), none
> open.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
