CAMEL-23938: add Camel-managed session lock renewal for session-enabled Service Bus consumers - #25001
Conversation
…ed Service Bus consumers Session-enabled Azure Service Bus consumers were excluded from the Camel-managed lock renewal added in CAMEL-23937, because session locks must be renewed via renewSessionLock() on a session-bound receiver rather than renewMessageLock(). This adds a SessionLockRenewer that: - creates session-bound async receivers on demand via acceptSession(sessionId) - renews session locks with renewSessionLock() as they approach expiry - reference-counts in-flight exchanges per session and closes the session receiver once the last exchange for a session completes - honours maxAutoLockRenewDuration per exchange Also adds ServiceBusClientFactory.createServiceBusSessionReceiverAsyncClient(), AssertJ test coverage, and an upgrade-guide entry. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com>
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Claude Code on behalf of gnodet
Nice extension — session lock renewal completes the picture
Good follow-up to CAMEL-23937. The architecture mirrors the existing LockRenewer cleanly, with the added complexity of session-bound receivers and reference counting properly justified by the Azure SDK's session lock API requirements. Tests are thorough (7 new tests covering the key scenarios). Upgrade guide updated.
One bug to fix
Session receiver leak when maxRenewDuration is exceeded or renewSessionLock() fails
In SessionLockRenewer.run(), two code paths remove entries from the entries map without calling releaseSessionReceiver():
- Max duration exceeded (line ~178):
entries.remove(exchangeId);
continue;- Renewal error handler (line ~199):
entries.remove(exchangeId);When run() removes the entry first, the SynchronizationAdapter.onComplete/onFailure callback later gets null from entries.remove() and skips releaseSessionReceiver(). This means the session receiver's inFlightCount is never decremented, so the session receiver is never closed — a resource leak.
The existing LockRenewer has the same entries.remove() pattern, but it doesn't matter there because it uses a single shared client with no reference counting. SessionLockRenewer has reference-counted SessionReceiverHolders that need explicit release.
Fix: call releaseSessionReceiver(entry.sessionId) alongside each entries.remove(exchangeId) in run():
// max duration exceeded
entries.remove(exchangeId);
releaseSessionReceiver(entry.sessionId);
continue;// error handler
error -> {
LOG.warn(...);
entries.remove(exchangeId);
releaseSessionReceiver(entry.sessionId);
}Minor observation (non-blocking)
The acquireSessionReceiver fast path (outside the synchronized block) can theoretically race with releaseSessionReceiver: thread A decrements to 0 and is about to remove + close, while thread B reads the holder, increments to 1, and returns the soon-to-be-closed client. In practice this requires a new message for the same session arriving at the exact instant the last exchange completes — unlikely but possible under load. The synchronized block only guards creation, not the acquire-vs-release interplay. Worth noting for a future hardening pass, but not a blocker for this PR.
Everything else looks good
ServiceBusClientFactory.createServiceBusSessionReceiverAsyncClient()correctly mirrors the receiver client setup withsessionReceiver()builder,disableAutoComplete(), andmaxAutoLockRenewDuration(Duration.ZERO)doStop()anddoShutdown()properly clean up both session and non-session resources (consistent with existing pattern)Boolean.TRUE.equals(isSessionEnabled())is null-safe ✓- Test coverage: session receiver reuse, reference counting lifecycle, stop/start cycle recreation — all the important scenarios
Address review feedback from gnodet: - Fix session receiver leak: SessionLockRenewer.run() removed entries from the tracking map without releasing the reference-counted session receiver on the max-renew-duration-exceeded and renewal-failure paths. The later completion callback then saw a null entry and skipped release, so inFlightCount never reached zero and the session receiver was never closed. run() now calls releaseSessionReceiver() when it wins the entries.remove(). - Harden acquire/release race: both acquireSessionReceiver() and releaseSessionReceiver() are now guarded by the same lock so the increment/create and decrement/close operations are atomic relative to each other, preventing a newly arriving message from incrementing a holder that a concurrently completing exchange is about to close. - Add test covering session receiver release on renewal failure. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com>
|
Thanks for the thorough review, @gnodet! Session receiver leak (fixed) — you're right. Acquire/release race (hardened) — also addressed. Added a test ( Pushed in b2042a5. Claude Opus 4.8 on behalf of atiaomar1978-hub |
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Clean follow-up to CAMEL-23937 that fills the explicitly-documented gap for session-enabled consumers.
The SessionLockRenewer correctly follows the patterns established by the LockRenewer in CAMEL-23937:
- Same
PeriodTaskSchedulerintegration andmaxRenewDurationhonoring - Same
SynchronizationAdapterfor exchange completion cleanup - Cleanup in both
doStop()anddoShutdown()matches the existing defensive pattern
Observations (non-blocking):
-
acceptSession(sessionId).block()inacquireSessionReceiver— synchronous blocking call inside theServiceBusProcessorClientcallback. The Azure SDK dispatches this callback on its own thread pool (not reactor threads), so.block()is safe. The first message for each unique session will block briefly on theacceptSessionhandshake, serialized throughsessionReceiverCreationLock; subsequent messages for the same session hit the fast path (cached receiver). Under extreme load with many distinct sessions arriving concurrently this could become a bottleneck, but unlikely in practice. -
Reference-counting is correctly synchronized —
acquireSessionReceiver/releaseSessionReceiverare both guarded by the same lock, preventing the race where an arriving message increments a holder that a concurrently completing exchange is about to close. -
Renewal failure handling is correct — when
renewSessionLock()fails, only the affected exchange's entry is removed and the ref count decremented. Other exchanges on the same session keep their entries and the session receiver stays open. -
Test coverage is thorough — 8 new tests covering creation conditions, exchange tracking/cleanup, renewal invocation, failure handling, session receiver reuse, reference counting, and stop/start lifecycle.
-
Upgrade guide — correctly updates the existing CAMEL-23937 entry rather than creating a separate section.
This review was generated by an AI agent (Claude Code on behalf of davsclaus) 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:
|
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 10 tested, 28 compile-only — current: 9 all testedMaveniverse Scalpel detected 38 affected modules (current approach: 9).
|
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Fundamental design issue with session lock contention
The SessionLockRenewer architecture has a critical flaw: acceptSession(sessionId).block() will always fail when called from inside the processMessage callback because the ServiceBusProcessorClient already holds an exclusive session lock on that session via its internal session receiver.
Azure Service Bus enforces exclusive session locks — only one receiver can hold a lock on a session at a time (Microsoft docs). A second acceptSession() call for the same session will receive a SessionCannotBeLockedException after exhausting retry attempts (~4 minutes with default AmqpRetryOptions). The catch block silently swallows this exception and returns null, causing add() to skip lock renewal registration — the feature silently does nothing in production.
The unit tests mask this because they mock acceptSession() to return immediately: when(sessionRenewalClient.acceptSession(any())).thenReturn(Mono.just(sessionBoundReceiver)).
Additional confirmed findings
-
The feature may be unnecessary: The Azure SDK documentation for
ServiceBusSessionProcessorClientBuilder.maxAutoLockRenewDurationstates that session lock renewal runs as a background task independent of callback duration — unlike per-message locks. The existingcreateServiceBusSessionProcessorClient()already configuresmaxAutoLockRenewDurationon the session processor builder. -
Redundant API calls (medium):
lockedUntilis tracked per-exchange rather than per-session. After a successfulrenewSessionLock(), only the triggering exchange's entry is updated. Other entries for the same session retain the stale value, producing N redundant API calls per renewal cycle. -
Missing timeout on
block()(medium):acceptSession(sessionId).block()has no explicit timeout. Under contention, this blocks theprocessMessagecallback thread for ~4 minutes before failing. -
cancel()doesn't clearentriesmap (low): Minor cleanup —cancel()clearssessionReceiversbut notentries. Low impact since the renewer is set to null immediately after.
Recommendation
This needs to be verified against a real Azure Service Bus session-enabled queue before merging. Alternative approaches to consider:
- Rely on the SDK's built-in session lock renewal via
maxAutoLockRenewDuration(may already be sufficient) - Switch from
ServiceBusProcessorClientto a lower-levelServiceBusSessionReceiverAsyncClientfor session consumers so Camel controls the session lifecycle directly
The architecture (reference counting, synchronized acquire/release, cleanup in doStop/doShutdown) is well-structured and the tests are thorough — the issue is with the fundamental approach, not the implementation quality.
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
Converting to draft — this PR is likely unnecessary, and the current approach cannot work as designedThank you @gnodet and @davsclaus for the careful reviews. After digging into the Azure Service Bus SDK ( 1. The
|
|
okay lets close this and resolve that jira as not a problem |
3 participants
CAMEL-23938: Camel-managed session lock renewal for session-enabled Service Bus consumers
Follow-up from CAMEL-23937.
Problem
CAMEL-23937 added Camel-managed lock renewal for non-session Azure Service Bus consumers using
ServiceBusReceiverAsyncClient.renewMessageLock(). Session-enabled consumers were explicitly excluded, because session locks require a different API:renewSessionLock()instead ofrenewMessageLock()acceptSession(sessionId))As a result, session-enabled consumers in async routes still suffered from silent session-lock expiry mid-processing.
Solution
Added a
SessionLockRenewerinServiceBusConsumerthat:ServiceBusSessionReceiverAsyncClient.acceptSession(sessionId)renewSessionLock()as they approach expirymaxAutoLockRenewDurationper exchangeAlso added
ServiceBusClientFactory.createServiceBusSessionReceiverAsyncClient().The behavior activates automatically when: no custom
processorClient,PEEK_LOCKmode,maxAutoLockRenewDuration > 0, and session mode is enabled.Tests
Added AssertJ test coverage in
ServiceBusConsumerTest:renewSessionLock()invoked when the lock is expiringmvn test -pl components/camel-azure/camel-azure-servicebuspasses.Docs
Updated the 4.22 upgrade guide entry to cover session lock renewal.
Claude Opus 4.8 on behalf of atiaomar1978-hub