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

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

shashank created CAMEL-25582:
--------------------------------

             Summary: 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


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)