Repository navigation
CAMEL-25841: camel-xmpp - do not hand out the endpoint of another participant or user - #28262
allthingssecurity wants to merge 5 commits into
Conversation
…ticipant or user XmppComponent kept the endpoints it created in a map keyed by scheme, host, port and query only, and returned a cached endpoint for any URI with the same key. So xmpp://alice@host/bob@host and xmpp://alice@host/carol@host were the same endpoint: messages for carol were sent to bob. Endpoints of different users on the same server were also shared, so messages went out from the first user's account. The component now creates a new endpoint for each URI, as other components do; the CamelContext already reuses the endpoint of the same URI and stops the endpoints it holds. The endpoints of one account must still share its connection: a server keeps one session per account and resource, and closes the older session with a conflict error when another one logs in (Smack does not reconnect after that). The component now keeps one connection per account (host, port, service name, user, password, resource, login, connection configuration), used by the consumers and producers of all its endpoints, and closes it when the last of them stops. Each endpoint keeps its own participant, room and other settings. A consumer that stops removes its listeners from the shared connection. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
gnodet-bot
left a comment
There was a problem hiding this comment.
The shared-connection model is the right approach — one XMPP session per account, shared across endpoints. The tests are solid: FakeXmppServer is a real in-process XMPP server, and the test scenarios directly reproduce the described failure modes. One correctness gap from the shared-connection model needs to be addressed before merge.
[HIGH] Concurrent reconnect race on shared connection
XmppGroupChatProducer.reconnect() and XmppPrivateChatProducer.reconnect() each acquire their own BaseService.lock — a different lock instance per producer. The inline reconnect blocks in XmppDirectProducer.process() and XmppPubSubProducer.process() hold no lock at all. When multiple producers of the same account all detect disconnection simultaneously, they race to call connection.connect() on the same shared XMPPTCPConnection.
Before this PR, each producer had its own connection, so concurrent reconnect() calls were safe by construction. After this PR, that guarantee is gone.
Smack 4.4.8 serializes connect() internally and throws AlreadyConnectedException (a SmackException) to the second caller. In XmppDirectProducer.process() and XmppPubSubProducer.process(), only XMPPException is caught — SmackException propagates as an unhandled exception, failing that exchange. In XmppGroupChatProducer and XmppPrivateChatProducer, the reconnect() method declares throws SmackException and the caller wraps it in RuntimeExchangeException.
Note: the companion PR #28263 (CAMEL-25842) handles AlreadyConnectedException in the consumer's connection monitor — the same fix is needed in the producers' reconnect paths.
Fix: Add a reconnectIfNeeded(XmppEndpoint) method to XmppComponent that holds SharedConnection.lock around the isConnected() check and connect() call, then have all four producers delegate through it:
// XmppComponent
void reconnectIfNeeded(XmppEndpoint endpoint)
throws InterruptedException, IOException, SmackException, XMPPException {
List<Object> key = endpoint.getConnectionKey();
SharedConnection shared;
synchronized (connections) {
shared = connections.get(key);
}
if (shared == null) return;
shared.lock.lock();
try {
if (shared.connection != null && !shared.connection.isConnected()) {
shared.connection = endpoint.createConnection();
}
} finally {
shared.lock.unlock();
}
}Then in each producer, replace connection.connect() with endpoint.getComponent().reconnectIfNeeded(endpoint).
generated by an AI agent, Hermès on behalf of @gnodet.
…tion once per account From gnodet-bot's review of apache#28262: now that the consumers and producers of an account share one connection, producers that find it disconnected at the same time each called connect() on it under their own lock (or none), and Smack 4.4.8 throws AlreadyConnectedException to the second caller, which failed that exchange. The producers now reconnect through XmppEndpoint.reconnect(), which holds the account's lock and treats AlreadyConnectedException as "another user of the account connected it". The javadoc of the connection key and the component page say that endpoints that set connectionConfig share a connection only when they refer to the same ConnectionConfiguration instance (it has no equals()). New tests: producersOfTheAccountReconnectTheSharedConnectionConcurrently (40 sends from 8 threads right after the server dropped the session; fails 3 of 3 runs without the change with "AlreadyConnectedException: Client is already connected") and reconnectAcceptsAConnectionThatAnotherUserOfTheAccountConnected. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
On the bot's points: both done in e980750.
While testing this I saw that the producers' reconnect only calls Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
There was a problem hiding this comment.
The concurrent reconnect race in XmppConsumer.checkConnection() is not yet addressed.
The PR correctly routes reconnect through endpoint.reconnect(connection) (which acquires the SharedConnection lock and handles AlreadyConnectedException) in all four producers. However, XmppConsumer.checkConnection() (line 154–164) still calls connection.connect() directly:
private void checkConnection() throws Exception {
if (!connection.isConnected()) {
LOG.info("Attempting to reconnect to: {}", XmppEndpoint.getConnectionMessage(connection));
try {
connection.connect(); // ← bypasses SharedConnection.lock
LOG.debug("Successfully connected to XMPP server through: {}", connection);
} catch (SmackException e) {
LOG.warn("Connection to XMPP server failed. Will try to reconnect later again.", e); // ← misleading: AlreadyConnectedException hits this branch
}
}
}This is called from startRobustConnectionMonitor() via a scheduleAtFixedRate executor. With shared connections, multiple consumers on the same account each have their own scheduled executor all calling checkConnection() concurrently. When the shared connection drops:
- All consumers fire
checkConnection()at once - The first one acquires the underlying XMPP connection and calls
connect() - The second one (racing) also passes
!connection.isConnected()and callsconnection.connect() - Smack throws
AlreadyConnectedException(aSmackExceptionsubclass) → caught at line 160 → logs a misleadingWARN("Connection to XMPP server failed")when it actually succeeded
The fix is the same pattern already applied to the producers: route through endpoint.reconnect(connection) instead of connection.connect() directly:
private void checkConnection() throws Exception {
if (!connection.isConnected()) {
LOG.info("Attempting to reconnect to: {}", XmppEndpoint.getConnectionMessage(connection));
try {
endpoint.reconnect(connection);
LOG.debug("Successfully connected to XMPP server through: {}", connection);
} catch (SmackException e) {
LOG.warn("Connection to XMPP server failed. Will try to reconnect later again.", e);
}
}
}endpoint.reconnect(connection) (→ XmppComponent.reconnect()) acquires the SharedConnection.lock, does a double-check if (!connection.isConnected()) under the lock, and catches AlreadyConnectedException at DEBUG level — exactly the semantics needed here.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
|
On the bot's point about Once both are in, I'll do one small follow-up that moves the consumer onto Claude Code on behalf of allthingssecurity |
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Shashank, the endpoint cache fix is right: the key left out the participant and the user, and the CamelContext already keeps one endpoint per URI. Sharing one connection per account is the right follow-on, since Smack 4.4.8's ReconnectionManager does not reconnect after a conflict stream error (I checked the jar).
- Blocking:
acquireConnectionreplaces the account's connection with a new one when it is not connected (XmppComponent.java:125). The consumers and producers that acquired it earlier keep the old object. So when anything acquires during an outage (a route that starts, atoDproducer that sends its first message, the consumer's delayed start), the account ends up with two connection objects that have the same JID and resource. Smack'sReconnectionManager(and, with #28263, the consumer's monitor) logs the old one in again, and the server closes one of the two withconflict. That is the case this PR sets out to avoid. The old object is also never disconnected:releaseConnectiononly disconnectsshared.connection, so the old connection and its session stay open after the last route stops. You kept the existing object inreconnectfor this reason;acquireConnectionshould do the same: create a connection only when there is none, otherwise connect (and log in) the existing object under the lock. - Not blocking, but I'd fix it here: the consumer and the group chat producer call
muc.leave()/chat.leave()indoStopbefore they release the connection.leave()throws when the connection is down or when Smack no longer has us in the room (MucNotJoinedException). Then the release is skipped, the account's user count never gets to zero, and the connection stays open (and Smack keeps reconnecting it) until the component stops. In the consumer the chat listener also stays registered. Please release in afinally. - Not blocking: please add a short entry to
camel-4x-upgrade-guide-4_24.adoc. Users will notice that URIs that differ only in participant or user are now separate endpoints. All endpoints of one account now share one connection: one session instead of one per endpoint, closed when the last route of the account stops, and shared only through the sameConnectionConfigurationinstance whenconnectionConfigis set. The publicXmppEndpoint.createConnection()no longer returns the endpoint's connection; it creates a new one on every call, and the endpoint no longer disconnects anything when it stops. - Not blocking:
producersOfTheAccountReconnectTheSharedConnectionConcurrentlyonly checks that no exchange fails. The messages sent after the reconnect are not delivered, because the producers do not log in again (the same on main). The follow-up you mention is fine, but please file a JIRA for it. - The open thread on
connectionConfigidentity is covered by the javadoc ofgetConnectionKeyand the new "Connections" section, so it can be resolved.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
… instead of replacing it, and release it when leaving a room fails Per review: - acquireConnection creates a connection only when the account has none; a connection that is not connected is connected and logged in again under the account's lock (connectAndLogin, also used by reconnect, which now logs in again too), so the account never has two connection objects with the same JID and resource. - Releasing the last user disconnects the connection also when it is not connected, which stops Smack's ReconnectionManager from connecting it again. - XmppConsumer and XmppGroupChatProducer remove their listeners and release the connection in a finally after leaving their room, so a failing leave() does not keep the account's connection open. - Upgrade guide 4.24: separate endpoints per participant and user, one connection per account, and XmppEndpoint.createConnection() creating a new connection on every call. - Tests: acquire during an outage gets the same connection logged in again; a consumer and a group chat producer that cannot leave their room release the connection; delivery after the concurrent reconnects; the fake connection also throws AlreadyLoggedInException. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Thanks Claus. Pushed 81f6fe2 on top:
The branch still merges cleanly with #28263 and with main ( Claude Code on behalf of allthingssecurity |
|
🧪 CI tested the following changed modules:
🔬 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, 6.9s total)Total reactor time: 6.9s
Top 20 slowest modules:
|
gnodet-bot
left a comment
There was a problem hiding this comment.
Re-reviewed the latest commit — all previously raised concerns are now addressed:
- ✅
acquireConnectionsecond-connection conflict (davsclaus): fixed — the method now reusesSharedConnection.connectionand callsconnectAndLogininstead of creating a conflicting second connection. - ✅
muc.leave()connection leak (davsclaus): fixed — bothXmppConsumer.doStop()andXmppGroupChatProducer.doStop()now wrap leave calls intry/finallythat callsreleaseConnection(). - ✅ Producer reconnect race (prior review): the
XmppGroupChatProducer.reconnect()path correctly acquires the producer lock before delegating to endpoint, maintaining consistent lock ordering.
The shared connection design is well-implemented: SharedConnection.connection is volatile (correct for cross-thread visibility), lock ordering is consistent (producer.lock → shared.lock → endpoint.lock), and the new tests (XmppComponentEndpointPerUriTest, XmppSharedConnectionTest) provide solid coverage. The checkConnection() deferral to #28263 is a reasonable split.
LGTM ✅
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Shashank, this addresses everything from my last review:
acquireConnectionnow connects and logs in the account's existing connection instead of replacing it, so there is only ever one connection object per account andreleaseConnectioncloses it. I checked against Smack 4.4.8:login()reuses the previous user, password and resource, andReconnectionManagertoleratesAlreadyConnectedException/AlreadyLoggedInException, so racing with it is safe.- The release in a
finallyafterleave()looks good, and the two new tests cover it. - The upgrade guide entry reads well.
- Thanks for making the concurrent reconnect test check delivery. Please file the follow-up JIRA (send between connect and login, rejoining the room after a reconnect, the consumer monitor reconnecting under the account's lock) and link it here.
Small nit, optional: in XmppConsumer.doStop, releaseConnection() is called both in the finally and right after the if. It's harmless because the method is idempotent, but a single call in an outer finally would be easier to read.
Approving, to merge once CI is green.
Claude Code on behalf of @davsclaus.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
…gain Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… on the test classpath
smack-core 4.4.8 asks for jxmpp-core and jxmpp-jid in the range [1.0.0,1.0.999],
while camel-xmpp pins jxmpp-jid to ${jxmpp-version} (1.1.0). The default resolver
then uses jxmpp 1.0.3 everywhere, but the CI build (mvnd) left jxmpp-util-cache off
the test classpath of apache#28263, so Smack failed to initialize (NoClassDefFoundError:
org/jxmpp/util/cache/Cache). This PR adds Smack tests too; declaring it explicitly
keeps it there for any resolver.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Note: 252c279 declares Claude Code on behalf of allthingssecurity |
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Shashank, the javadoc of reconnect now matches what connectAndLogin does, in both XmppEndpoint and XmppComponent. Declaring jxmpp-util-cache here too makes sense, since this PR adds Smack tests that run on CI as well. It merges cleanly with #28263, which adds the same pom lines.
A reminder: please file the follow-up JIRA (send between connect and login, entering the group chat room again after a reconnect, the consumer monitor reconnecting under the account's lock) and link it here.
Approving, to merge once CI is green.
Claude Code on behalf of @davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
Follow-up JIRA (producers sending before the login, the group chat producer entering its room again, the consumer monitor reconnecting under the account's lock): CAMEL-25961. Claude Code on behalf of allthingssecurity |
Description
CAMEL-25841
XmppComponentkept the endpoints it created in a map keyed by scheme, host, port and query, and returned the cached endpoint for any URI with the same key. The participant (the path) and the user are not part of the key, soxmpp://alice@host/bob@host?password=xandxmpp://alice@host/carol@host?password=xwere the same endpoint: messages for carol went to bob. Two users on the same server shared an endpoint too, so messages went out from the first user's account. This happens with static endpoints in two routes as well as with toD.Separate endpoints alone are not enough: each endpoint opened its own connection, and the endpoints of one account (
xmpp://bot@host/alice,xmpp://bot@host/bob) log in with the same JID and the same resource (Camelby default). An XMPP server keeps one session per account and resource and closes the older one with aconflictstream error, and Smack 4.4.8'sReconnectionManagerdoes not reconnect afterconflict, so the usual "one account, several participants" setup would lose an endpoint for good. A distinct resource per endpoint is not a fix either: a message to the bare JID goes to one resource only.Changes:
XmppComponent: remove the endpoint cache. The component creates a new endpoint for each URI, like other components. The CamelContext already reuses the endpoint of the same URI and stops the endpoints it holds.XmppComponent: keep one connection per account instead, keyed by the settings that make up the connection (host, port, service name, user, password, resource, login, connection configuration), shared by the consumers and producers of all endpoints of that account, and closed when the last of them stops (and at component stop). A connection that is no longer connected when a consumer or producer acquires it is connected and logged in again, not replaced: the others keep using that object (and Smack'sReconnectionManagerkeeps reconnecting it), so a second object would log in the same JID and resource and the server would close one of them (conflict). Releasing the last user also disconnects a connection that is no longer connected, which stops Smack'sReconnectionManagerfrom connecting it again.XmppEndpoint.acquireConnection(user)/releaseConnection(user): what the consumers and producers now use instead ofcreateConnection(), which only creates a new connection. Each endpoint keeps its own participant, room, nickname and other settings.XmppConsumerand the producers release the connection when they stop instead of disconnecting it (it may serve other endpoints). The consumer also removes its listeners from the connection, so a restarted consumer does not receive each message twice. The consumer and the group chat producer do this in afinallyafter leaving their room, so aleave()that fails (the connection is down, or Smack no longer has us in the room) does not keep the account's connection open. The direct and pubsub producers now release their connection on stop too (before, only the endpoint's stop closed it).XmppEndpoint.reconnect(connection), which holds the account's lock: producers that found it disconnected at the same time each calledconnect(), and Smack throwsAlreadyConnectedExceptionto the second caller, which failed that exchange.AlreadyConnectedExceptionthere now only means that another user of the account connected it (review follow-up e980750). The reconnect also logs the connection in again (Smack'slogin(), with the user, password and resource it logged in with before;AlreadyLoggedInExceptiontolerated): before, the producers only connected it, also on main, and the server delivered nothing they sent afterwards.pom.xml:jxmpp-util-cacheis declared next tojxmpp-jid(same${jxmpp-version}), as in CAMEL-25842: camel-xmpp - the consumer logs in and enters its room again after its connection was lost #28263: Smack needs it, and it comes in only through jxmpp-core, which Smack 4.4.8 asks for in the range[1.0.0,1.0.999]; a CI build left it off the test classpath of the Smack tests.connectionConfigshare a connection only when they refer to the sameConnectionConfigurationinstance, since that class has noequals(); the component page has a short "Connections" section that says so.The connection of an account is still closed when its last route stops, as before for a single endpoint. No option or default changes. Users can notice the separate endpoints, the shared connection and the changed
XmppEndpoint.createConnection()(it now creates a new connection on every call, and an endpoint no longer disconnects anything when it stops), so there is an entry in the 4.24 upgrade guide.Not in this PR (follow-up JIRA): a producer only checks
isConnected()before it sends, so a send that comes while the connection is connected but not yet logged in again goes out before the login; the group chat producer does not enter its room again after a reconnect; the consumer's connection monitor (#28263) reconnects without the account's lock.Tests:
XmppComponentEndpointPerUriTest(3 tests, no XMPP server needed).endpointPerParticipantandendpointPerUserfail without the change (expected: <carol@localhost> but was: <bob@localhost>,expected: <dave> but was: <alice>).sameUriSameEndpointpasses before and after.XmppSharedConnectionTest(8 tests) withFakeXmppServer, a small in-process XMPP server (plain TCP, SASL PLAIN and ANONYMOUS, resource binding, room presence) that, like a real server, closes the older session with aconflictstream error when another session binds the same JID and resource. The real Smack client connects to it.endpointsOfOneAccountShareItsConnection: two endpoints ofbotwith participants alice and carol each send to their own participant over one session. On main carol's message goes to alice (expected: <[hello alice]> but was: <[hello alice, hello carol]>); with a connection per endpoint, the second login closes the first session (expected: <[hello alice]> but was: <[]>).stoppingOneEndpointKeepsTheConnectionOfTheOther: stopping and removing alice's endpoint keeps the session for carol's; the session is closed when the last route of the account stops. With a connection per endpoint it fails (The account logged in more than once ==> expected: <1> but was: <2>).consumerRestartedOnASharedConnectionReceivesEachMessageOnce: a consumer restarted while a producer keeps the connection receives each message once (without removing its listener on stop:expected: <2> but was: <3>).producersOfTheAccountReconnectTheSharedConnectionConcurrently: 40 sends from 8 threads right after the server dropped the session. Without the reconnect under the account's lock it fails 3 of 3 runs withAlreadyConnectedException: Client is already connected.producersOfTheAccountReconnectTheSharedConnectionConcurrentlyalso checks that a message sent after the reconnects is delivered (fails when the reconnect does not log in).reconnectAcceptsAConnectionThatAnotherUserOfTheAccountConnected: a connection whoseconnect()andlogin()throwAlreadyConnectedExceptionandAlreadyLoggedInException, as Smack does for the second caller, is accepted.connectionAcquiredDuringAnOutageIsTheConnectionOfTheAccount: the server drops the session, something of the account acquires the connection during the outage and gets the same object, logged in again; the producer that held it delivers again, with noconflict. With a new connection per acquire it fails (A second connection of the account was created).consumerThatCannotLeaveItsRoomReleasesTheConnectionandgroupChatProducerThatCannotLeaveItsRoomReleasesTheConnection: the connection is dropped, the route stops (leave()throwsNotConnectedException), and the account has no users left. Without thefinallythey fail (expected: <0> but was: <1>).-Daether.dependencyCollector.impl=bf. The integration tests were not run.Unchanged: a consumer receives every chat message of its account, whatever its participant, as before; a group chat producer and a consumer of the same room on one account share Smack's
MultiUserChatof that room, as they did before when they used the same endpoint.How it was found: a sweep for components that keep endpoints or clients in their own maps, then the test above.
Target
mainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.(I built and tested camel-xmpp. No option or generated metadata changed. I did not run the full root build.)
AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a
Co-Authored-Bytrailer.Claude Code on behalf of allthingssecurity
🤖 Generated with Claude Code