CAMEL-24569: Support Salesforce Streaming API disconnect messages - #25908
Conversation
Handle server-initiated /meta/disconnect messages introduced in Salesforce Streaming API 64.0 and reconnect to keep subscriptions active.
…ssing a disconnect message - Wait for CometD to reach the disconnected state before reconnecting. - Use the managed executor and verify that channels are resubscribed only once.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Reviewed against the project's contribution rules and build/test conventions (not a substitute for CodeRabbit/Sourcery/SonarCloud or other specialized static analysis).
Verification performed: built camel-salesforce-component from this branch (mvn install -DskipTests — clean). Ran the project's own formatter (mvn formatter:format impsort:sort) and it produced zero diff — the large amount of line-rewrapping in this diff looked at first glance like it might be from a different formatter/IDE config than the project's, but it's actually exactly what Camel's own formatter produces, so it's legitimate fallout of the mandated mvn clean install -DskipTests step, not scope drift. Ran SubscriptionHelperManualIT (a self-contained stub-server test — it only executes under the integration Maven profile, not default CI, despite the name) directly via surefire: the new shouldResubscribeOnDisconnectMessage test passes. One other test in the same class, shouldResubscribeOnSubscriptionFailure, fails — I confirmed by running the identical test against unmodified main that this is a pre-existing failure unrelated to this PR (an NPE from an unstubbed mock), not a regression introduced here.
I also compared the two commits: the first pass used a while loop with Thread.sleep() for reconnect backoff, and the second commit ("Improve Salesforce component to be state-safe...") replaced it with a waitFor(..., DISCONNECTED) check that delegates back into the existing handshake/backoff machinery. That's a good self-correction, and it fits well with this file's history of hardening against races (CAMEL-23391, CAMEL-24160/24161, CAMEL-24272) — appreciate that the reconnect logic reuses the established resubscribe-on-handshake path rather than reimplementing it.
No blocking issues found. Two non-blocking questions below (see inline comments), plus one process note: JIRA CAMEL-24569 is currently Unassigned/Open — per project convention it should be assigned to the contributor and moved to "In Progress" before merge.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Solid PR — the disconnect handling and reconnection logic are well-designed with proper thread safety.
The AtomicBoolean reconnection guard prevents concurrent reconnect attempts, the volatile upgrades on client and taskExecutor combined with local-reference captures (final BayeuxClient disconnectedClient = client) are the correct idiom for safe publication across threads, and the shutdown sequence in doStop() correctly removes the disconnect listener before calling client.disconnect() so that client-initiated disconnects during shutdown won't trigger reconnection.
The test follows existing patterns in SubscriptionHelperManualIT and validates the complete disconnect → reconnect → resubscribe lifecycle, including the await().during(1, SECONDS) assertion that catches any double-reconnection races.
Two minor observations (neither blocking):
-
In
createDisconnectListener(), the catch block only handlesRejectedExecutionException. Ifexecutor.execute()threw any other unchecked exception, thereconnectingflag would remaintrue, permanently disabling reconnection. In practiceRejectedExecutionExceptionis the only runtime exceptionScheduledThreadPoolExecutor.execute()can throw, so this is fine — just noting for completeness. -
reconnectAfterDisconnect()callsdisconnectedClient.waitFor(waitMs, DISCONNECTED)which blocks thetaskExecutorthread for up to 120 seconds. This is an acceptable trade-off given the existing design, but worth noting for future reference.
📋 PR Metadata
| Aspect | Current | Suggested |
|---|---|---|
| Labels | components |
+ bug |
| Milestone | (none) | 4.23.0 |
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
Claude Code on behalf of Guillaume Nodet
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 11 tested, 27 compile-only — current: 11 all testedMaveniverse Scalpel detected 38 affected modules (current approach: 11).
|
…tate in the http worker thread instead of a single-threaded executor service not to block the latter. Also, disconnect timeout has been reduced to 10 seconds
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Well-implemented handling of Salesforce Streaming API 64.0 server-initiated disconnect messages. The core design is solid: the AtomicBoolean guard prevents concurrent reconnects, volatile fields ensure visibility across threads, and isStoppingOrStopped() checks at multiple points properly handle the shutdown race. Capturing the client reference locally as disconnectedClient and comparing against the field later is a nice pattern to detect stop/restart.
A few observations (none blocking):
-
Formatting noise: The PR mixes ~70% formatting-only changes (re-wrapping lambdas,
String.formatarguments, method signatures) with the functional additions (disconnect listener, reconnect method, volatile fields). This makes the functional diff harder to isolate and increases merge conflict risk. Consider splitting formatting into a separate commit in future PRs. -
Exception handling in
createDisconnectListener(): OnlyRejectedExecutionExceptionis caught. Ifcomponent.getHttpClient()orgetWorkerPool()throws an unexpectedNPEor otherRuntimeExceptionduring partial shutdown, thereconnectingAtomicBoolean staystrue, permanently preventing future reconnect attempts. Afinallyblock to reset the flag would add resilience. -
Disconnect timeout: When
waitFor(DISCONNECT_TIMEOUT_SEC, DISCONNECTED)times out, the method logs a warning and returns without scheduling a retry — the subscription is silently lost. Consider making the timeout configurable or scheduling a delayed retry. -
Reconnect backoff:
disconnectedClient.handshake()is called without delay. If the server immediately disconnects again (misconfigured/failing server), this could produce a tight reconnect loop. A brief delay before the first reconnect after a server disconnect would add resilience.
The test follows the established StubServer pattern and the disconnect message accurately reflects what Salesforce sends. The volatile additions to client and taskExecutor are appropriate corrections for pre-existing visibility gaps.
📋 PR Metadata
| Aspect | Current | Suggested |
|---|---|---|
| Milestone | (none) | 4.23.0 |
LGTM ✅
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
Claude Code on behalf of Guillaume Nodet
4 participants
CAMEL-24569: Support Salesforce Streaming API disconnect messages
Description
Handle server-initiated /meta/disconnect messages introduced in Salesforce Streaming API 64.0 and reconnect to keep subscriptions active.
Target
mainbranch)Tracking
Apache Camel coding standards and style
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.