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)
