allthingssecurity opened a new pull request, #29199: URL: https://github.com/apache/camel/pull/29199
# CAMEL-26574: camel-netty - Client mode consumer: close the connection at stop, reconnect without blocking the event loop https://issues.apache.org/jira/browse/CAMEL-26574 A consumer with `clientMode=true` connects out to a server and reconnects when the connection is lost (`ClientModeTCPNettyServerBootstrapFactory`). Two defects: - **Stop with a shared worker group.** `stopServerBootstrap()` never closed the channel; it only shut down a worker group it had created itself. With `workerGroup=#bean` the stopped consumer kept its connection (data from the server was read and dropped), a reconnect scheduled before the stop, or one started when the server closed the connection afterwards, connected again, and a restart left two connections that both fed the route. - **The reconnect waited on an event loop.** The reconnect task called `openChannel`, which waits with `awaitUninterruptibly()`. When the new channel got the same loop (`workerCount=1`, or a shared group), Netty threw `BlockingOperationException`; the log said "Will attempt again" but nothing was scheduled, and the connect went on without a close listener, so the consumer never reconnected after the next loss. When two reconnect tasks of a shared group (two consumers, or the leftover reconnect of a stopped consumer) each connected a channel on the loop of the other, both loops blocked for good: every channel of the group hung, and a route start waiting for its connect never returned. ## Changes - `stopServerBootstrap` cancels the scheduled reconnect, closes the channel and a connect in progress, and waits for the close up to `shutdownTimeout` (as the server mode does; not when called on the channel's own event loop). - The reconnect no longer waits: a connect listener tracks the new channel (and adds its close listener) or schedules the next attempt. An exception in the task now schedules the next attempt, as its log message says. - Channel, connect in progress, scheduled task, a stopping flag and a generation per start are guarded by one lock. The close listener, the reconnect task and the connect listener act only when the consumer is not stopping and belongs to the current start; a connect that completes after the stop or a restart is closed. The initial connect at start is unchanged (it still waits on the lifecycle thread and fails the start with `reconnect=false`). ## Tests `NettyConsumerClientModeStopTest` (embedded Netty server on localhost): - `stopClosesTheConnectionWithASharedWorkerGroup` - `noReconnectAfterStopWithASharedWorkerGroup` - `stopWhileReconnectingWithASharedWorkerGroup`: a pipeline factory holds the event loop of the reconnect channel once it is registered, before its connect completes; the stop runs in that window - `restartHasOneConnectionWithASharedWorkerGroup` - `twoConsumersReconnectWithASharedWorkerGroup`: the server closes both connections at once, five times; both consumers reconnect and no third connection appears - `reconnectsWithOneWorkerThread` - controls that pass before and after: `reconnectsAfterTheServerDroppedTheConnectionWithASharedWorkerGroup`, `stopClosesTheConnectionWithItsOwnWorkerGroup` The first six fail without the change (`twoConsumers...` 20 of 20 runs). All eight passed 20 times in a row with it; camel-netty: 146 run, 0 failures, 9 skipped. Found with TLA+ models of the factory and of two consumers on a shared group of two loops (the old code leaves a stopped consumer connected and lets the two loops block on each other; the changed models hold). A hook-free loop of 200 stop/start cycles against a server that closes every connection after 0-3 ms: a connection after the stop in 105 of 110 cycles, then a hang on the loop deadlock; with the change 0 of 200. No upgrade guide entry: no option or default changes. Stopping a client-mode consumer now waits up to `shutdownTimeout` (100 ms by default) for its connection to close. Backports: the class is identical on `camel-4.23.x`, `camel-4.18.x` and `camel-4.14.x`. _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]
