[Sunghyun Jung (sunghyun1999)]

Summary

• Add isDisposed() liveness check before returning cached send/receive links from ReactorSession.openSendLinks / openReceiveLinks
• If a cached link is disposed (e.g. after silent AMQP detach by LB/NAT idle timeout), remove it from the cache and create a fresh link instead of returning the stale one
• Apply the same check in the ConcurrentHashMap.compute() lambda to handle race conditions

Problem

When an AMQP sender link is silently detached (e.g. by infrastructure idle timeout on Kubernetes), the dead link remains in ReactorSession.openSendLinks cache. createProducer() returns this stale link on every subsequent call without checking if it's still alive. Combined with RetryUtil.withRetry() re-subscribing to the same Mono chain (which reacquires the same dead link), this causes:

1. Every sendMessage() call hangs for tryTimeout (60s default)
2. All retries fail identically → 245.6s total timeout
3. The instance is permanently stuck until JVM restart
4. Other queues on the same connection continue to work (cache is keyed by entity path)

Production evidence: 10/10 failures over 25 hours on one queue, while another queue on the same AMQP connection succeeded at 12–33ms. A fresh pod sent to the same queue in 87.8ms. See #50503 for full details.

Fix

The fix is minimal and surgical — a single isDisposed() check at the cache lookup boundary:

ReactorSession.java — createProducer() (send link cache):

final AmqpSendLink existingLink = existing.getLink();
if (existingLink != null && existingLink.isDisposed()) {
    logger.atInfo().addKeyValue(LINK_NAME_KEY, linkName)
        .log("Cached send link is disposed. Removing to allow recreation.");
    removeLink(openSendLinks, linkName);
} else {
    return Mono.just(existingLink);  // existing behavior
}

Same pattern applied to createConsumer() (receive link cache) and both compute() lambdas.

Why this is safe

• isDisposed() is a simple AtomicBoolean.get() — no side effects, no blocking
• When disposed, removeLink() cleans up the cache entry (already existing method)
• Falls through to the existing link creation path — no new code paths introduced
• A non-disposed link behaves exactly as before — zero behavioral change for healthy links
• The compute() lambda check handles the race between the early lookup and the CAS

Test plan

• New unit test: createProducerRecreatesDisposedLink — verifies that after disposing a cached link, createProducer() returns a new instance instead of the stale one
• Existing createProducer and createProducerAgainAfterException tests continue to pass (CI)
• Manual verification: deploy on AKS with Service Bus, verify link recovery after idle timeout

Related

• Fixes [BUG] ServiceBusSenderClient permanently stuck when a single AMQP sender link is detached — dead link cached in ReactorSession.openSendLinks #50503
• Related to stalled PR Add tiered send/receive recovery to azure-core-amqp and azure-messaging-servicebus #48460 (tiered recovery) — this fix is orthogonal and much smaller in scope
• Related closed issues: Add tiered send recovery to match Go/.NET/Python/JS SDK parity — stale connection causes permanent sender stuck #48447, [BUG] Servicebus-Java ''Timeout on blocking read' when sending messages' #44688, [BUG] Message sending in ServiceBus queue occasionally fails with error "Cannot send a message when request response channel is disposed" or "Timeout on blocking read" #43877

🤖 Generated with Claude Code

[github-actions]

Thank you for your contribution Sunghyun Jung (@sunghyun1999)! We will review the pull request and get back to you soon.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).
31 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.

[Copilot]

🟡 Changes recommended

Critical receive-link race and cleanup issues remain, and regression-test setup and coverage need correction.

Get a fresh assessment by requesting another Copilot review.

Pull request overview

Adds liveness checks to recreate disposed cached AMQP send and receive links.

Changes:

• Detects disposed links during cache lookup and computation.
• Removes stale links and creates replacements.
• Adds a producer recreation regression test.

File summaries

File	Summary
sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java	Adds send/receive link liveness and replacement logic; receive-path race and cleanup issues remain.
sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSessionTest.java	Adds producer recreation coverage; test setup requires lifecycle fixes and receiver coverage.

Review details

Suppressed comments (3)

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java:536

• This is a TOCTOU race: after the lookup at line 525 observes the disposed existing, another caller can install a healthy replacement before this key-only removal runs. removeLink will then close that healthy link, and this call will create yet another link. Remove conditionally using the observed existing subscription and close it only when that conditional removal succeeds.

                removeLink(openSendLinks, linkName);

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java:565

• When the compute lambda replaces a disposed cached link, the old LinkSubscription is not closed or unsubscribed. Its endpoint-state callback can run after the replacement is installed and call removeLink(openSendLinks, linkName), removing and closing the new link; it also leaves the old subscription attached. Close the disposed subscription before replacing it so its callback cannot affect the new cache entry.

                                final AmqpSendLink link = existingLink.getLink();
                                if (link != null && link.isDisposed()) {
                                    logger.atInfo()
                                        .addKeyValue(LINK_NAME_KEY, linkName)
                                        .log("Cached send link is disposed. Creating a new send link.");

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorSessionTest.java:340

• The provider normally creates a new one-shot SendLinkHandler for each link, but this stub reuses sendLinkHandler for both senders. Disposing the first link terminates that handler, so the recreated ReactorSender can inherit a completed endpoint stream and be immediately disposed; the test may then pass while the replacement is not live. Return a distinct handler for the second link and assert it is not disposed.

        when(reactorHandlerProvider.createSendLinkHandler(ID, HOST, linkName, entityPath)).thenReturn(sendLinkHandler);

• Files reviewed: 2/2 changed files
• Comments generated: 4
• Review effort level: Lite

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.

[Sunghyun Jung (sunghyun1999)]

Hi team, gentle ping on this PR — it's been open for about 6 weeks now with all CI checks passing. Would appreciate a review when you get a chance. Thanks!