Repository navigation
CAMEL-25812: camel-milo - end idle client threads and run reconnects on a thread of their own - #28111
Conversation
…on a thread of their own Since CAMEL-25713, a client connection handled the outcome of its requests and ran its reconnects on a scheduled thread pool of its own with one thread per processor. The threads of a scheduled pool never end while the connection is open, so 30 connections on a machine with 8 processors kept 240 idle threads. With virtual threads, the pool had a single worker, so after awaited requests the routes of a connection ran one at a time: one connection with 32 callers and 10 ms of work after each awaited write fell from about 670 to 85 exchanges per second. A connection now handles the outcome of its write, method call, read and browse requests on a thread pool with up to one thread per processor, whose idle threads end after the keep-alive time of the default thread pool profile. With virtual threads and more than one processor, the pool starts a virtual thread for each task, and the same measurement gives about 2,100 to 2,300 exchanges per second. The pool stays bounded with platform threads, because when the connection is lost every request in flight fails and waits for the disconnect of the first failure: a pool that starts a thread for each of them ran out of threads with writes that are not awaited, and the connection was not rebuilt. The reconnects run one after the other on a single-thread scheduled executor of the connection, so a reconnect no longer waits behind the outcomes of failed requests. Disposing the connection cancels a reconnect that is still waiting, so the thread ends when the connection is closed. Neither executor is the shared scheduled executor of Milo, so a failed request still disconnects without waiting for a thread that only the disconnect could free. A request that completes after the connection is closed still fails with a RejectedExecutionException. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Clean concurrency refactor — splitting the single ScheduledExecutorService into a plain ExecutorService for request outcomes and a dedicated SingleThreadScheduledExecutorService for reconnects is the right approach. Lock ordering is correct, dispose() properly cancels the reconnect job under the lock, and the new test validates the fix.
One low-severity suggestion below.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
…n the test AvailablePortFinder.getNextAvailable() is deprecated. The test now holds the port as an AvailablePortFinder.Port extension, as AbstractMiloServerTest does for the server port, so it stays reserved while the test runs. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Previous finding is addressed. ✅
The deprecated AvailablePortFinder.getNextAvailable() call has been replaced by @RegisterExtension AvailablePortFinder.Port unreachablePort = AvailablePortFinder.find() — the canonical pattern that holds the port reservation for the duration of the test, matching how AbstractMiloServerTest manages the server port. LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
There was a problem hiding this comment.
Thanks @etzel-mes, this is the follow-up I asked for in #28049, and the measurements explain well why a cached pool was not the answer with platform threads.
What I checked:
newThreadPoolmerges the default profile, so theMiloClientpool gets the 60 s keep-alive, and core threads time out (the factory's default). With virtual threads,maxPoolSize > 1and a queue size above 0 give theBoundedExecutorServiceover a thread per task, and with one processor it stays one platform thread, as the PR and the upgrade guide say.Integer.MAX_VALUEinstead of-1is right:-1gives aSynchronousQueue, which rejects a task while every thread is busy.- The reconnects now run on their own single-thread scheduled executor. That serialises them and stops them waiting behind the failed request outcomes.
connect()still returns early when the manager is disposed. dispose()cancels a waiting reconnect. The scheduled pools usesetRemoveOnCancelPolicy(true), so the shutdown inclose()lets the thread end at once. The new test covers that.close()shuts down both executors, each one even if the other fails.- The upgrade-guide section of #28049 now describes both pools.
One non-blocking nit: triggerReconnect does not check disposed. A connect that fails, or a request that fails, while the connection is being closed can still schedule a reconnect after dispose() has cancelled the waiting one. That reconnect returns at once when it runs, because connect() checks disposed. But a scheduled pool keeps delayed tasks after shutdown(), so the MiloClientReconnect thread can stay up to 10 s after the close in that narrow window. An early if (this.disposed) return; in triggerReconnect would close that window. Fine as part of the separate fix you plan for the failure handler that does not check its connection.
I resolved the gnodet-bot thread: the test now holds the port as an AvailablePortFinder.Port extension.
CI is still running on the current head.
Claude Code on behalf of Claus Ibsen
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
|
🧪 CI tested the following changed modules:
🔁 1 test passed only after a retry on JDK 25 (1 retried attempt) Recovered flaky tests on JDK 25 (1)
🔬 Scalpel shadow comparison — Scalpel: 8 of 707 tested, 26 compile-only — current: 8 all testedMaveniverse Scalpel detected 8 affected modules (current approach: 8). Skip-tests mode would test 8 modules (2 direct + 7 downstream), skip tests for 26 (generated code, meta-modules) Modules Scalpel would test (8)
Modules with tests skipped (26)
All tested modules (35 modules, 3m 59s total)Total reactor time: 3m 59s
Top 20 slowest modules:
|
Description
CAMEL-25812, the follow-up to #28049 (CAMEL-25713) that @davsclaus suggested in #28049 (comment).
Since #28049, a
milo-clientormilo-browseconnection handles the outcome of its write, method call, read and browse requests, and runs its reconnects, on a scheduled thread pool of its own with one thread per processor. The threads of a scheduled pool never end while the connection is open, and with virtual threads Camel builds that pool with a single worker.The change:
MiloClientConnectionhandles the request outcomes on a thread pool fromExecutorServiceManager.newThreadPoolnamedMiloClient. It has up to one thread per processor, and its idle threads end after the keep-alive time of the default thread pool profile (60 s). With virtual threads and more than one processor, Camel builds it as a virtual thread per task. Its queue is not bounded (Integer.MAX_VALUE, because-1gives aSynchronousQueuethat rejects when every thread is busy), and the rejected policy staysAbort, so a request that completes after the connection is closed still fails with aRejectedExecutionException.newSingleThreadScheduledExecutornamedMiloClientReconnect. Its thread starts with the first reconnect, and the reconnects run one after the other; before, two connects could run at the same time. They no longer wait in the queue behind the outcomes of the failed requests.SubscriptionManagertakes the two executors; onlytriggerReconnectuses the scheduled one.dispose()cancels a reconnect that is still waiting, so theMiloClientReconnectthread ends when the connection is closed instead of up to 10 s later.close()shuts down theMiloClientpool even if shutting down the reconnect executor fails.MiloClientExecutorTestgets a test that a connection that cannot connect has aMiloClientReconnectthread, and that the thread ends within 5 s after the route stops (it fails without the cancel indispose()). The idle timeout and the virtual thread per task come from Camel's thread pool factory; they are covered by the measurements below, not by a unit test.Neither executor is Milo's shared scheduled executor, so a failed request still disconnects without waiting for a thread that only the disconnect could free;
MiloClientExecutorTestfrom #28049 still covers that. The values that a reconnect delivers to the consumers run their routes on the reconnect thread, and an awaited request from such a route completes on theMiloClientpool. On #28049, with one processor or with virtual threads, the reconnect and that completion shared the single worker of the pool.Because a reconnect no longer queues behind the outcomes of the failed requests, a failure that is handled after the reconnect can now dispose the new connection, which then reconnects once more. This is not new: a failure handler does not check that it belongs to the current connection. In the reconnect probe below it shows as one more connect with one processor and awaited writes (3 instead of 2), and the connection still recovers. I plan to fix that check separately, after #27956, which changes the same producer code.
I first tried a cached pool for the request outcomes, as suggested, but it does not work with platform threads. When the connection is lost, every request in flight fails, and each failure waits for the lock of the connection while the first failure disconnects. With four producers sending writes that are not awaited, and the server frozen for 30 s, the cached pool started about 4,080 threads until the JVM could not create more (
pthread_create failed (EAGAIN)), and the connection was never rebuilt. So with platform threads the pool keeps the bound of #28049 and only gains the idle timeout; with virtual threads, it starts a thread per task.Measured with host JVMs (Java 21, client limited with
-XX:ActiveProcessorCount=8, against a local Milo server, two runs each):MiloClientthreads 5 s / 65 s after the load(*) #28049 and this PR alternated run by run, three runs each, each started only when the 1-minute load of the host was below 6. Main is from the earlier matrix; its runs without work varied twofold on the busier host, so they are left out.
A reconnect probe froze the server JVM (
SIGSTOP, so the sockets stay open and nothing answers) for 30 s, while one connection had a consumer and four producers, and checked when the consumer received values again after the server resumed. This PR recovered in every run, 1–6 s after the resume: with 1 and 8 processors, with awaited writes and with writes that are not awaited, and with virtual threads (8 processors). Its peak was 135 platform threads. On main, with one processor and awaited writes, the probe also reproduced CAMEL-25713:milo-shared-scheduled-executor-0waited inOpcUaClient.disconnect()fromhandleConnectionFailureof a write, and the connection was not rebuilt.Against a real server, opc-plc 2.15.4 in Docker, a client with a consumer and a producer that does an awaited write every 2 s lost its traffic silently (an iptables DROP) for 30 s, while #28049 ran beside it on a second connection, positions alternating. This PR recovered in all 16 runs (6 with 1 processor, 6 with virtual threads and 2 processors, and 4 more with 1 processor on the final commit), 1–12 s after the traffic was restored, in the same range as #28049 in the same runs. With 1 processor it more often needed a third connect, the race described above.
mvn -Psourcecheck clean verifyincomponents/camel-milopasses (82 tests, 1 skipped; #28049 has 81), and the module build regenerates no files.@davsclaus @oscerd could you take a look?
Target
mainbranch)Tracking
Apache Camel coding standards and style
I checked that each commit in the pull request has a meaningful subject line and body.
I have run
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.Claude Code on behalf of etzel-mes