Fix memory leak in RequestResponseChannel by draining SendLinkHandler link credits (#47261)

[ksalazar-91]

Summary

Fixes a memory leak in RequestResponseChannel (azure-core-amqp), reported in #47261.

The channel's SendLinkHandler exposes link credits via getLinkCredits(), a unicast, unbounded-buffer sink (Sinks.many().unicast().onBackpressureBuffer()) that emits a value on every AMQP flow frame (onLinkFlow). ReactorSender subscribes to and drains this flux, but RequestResponseChannel — which backs all management / request-response operations (lock renewal, peek, schedule, session state, etc.) — only subscribed to the send link's endpoint states and never to its credits. With no subscriber, every emitted credit is retained in the sink's buffer for the lifetime of the channel.

Because the request-response channel is cached (RequestResponseChannelCache) and lives for the whole connection, a long-running, high-throughput client accumulates one buffered Integer per flow frame indefinitely, matching the reporter's heap analysis:

ManagementChannel -> RequestResponseChannelCache -> RequestResponseChannel -> SendLinkHandler

Fix

RequestResponseChannel now subscribes to (and thereby drains) sendLinkHandler.getLinkCredits(), registering the Disposable in its subscriptions composite so it is disposed on channel close. This mirrors the channel's existing endpoint-state / delivered-message subscriptions and how ReactorSender already drains credits. RequestResponseChannel correlates responses via unconfirmedSends and does not use AMQP link-credit flow control, so discarding the credit values is safe and behavior-preserving.

Reason

Steady, unbounded heap growth for long-lived Service Bus / AMQP clients under sustained load, eventually leading to OutOfMemoryError.

Fixes #47261.

Verification

• New regression test RequestResponseChannelTest#subscribesToSendLinkCreditsToPreventLeak asserts the channel subscribes to the credit flux and disposes the subscription on close.
• RequestResponseChannelTest (16/16) and SendLinkHandlerTest (7/7) pass; checkstyle and spotbugs are clean.
• Reproduced with a standalone program driving the real SendLinkHandler over 2,000,000 simulated flow frames: without the drain ~38 MB retained (all 2M credits replayed to the first subscriber); with the drain ~0 MB retained.

Notes for reviewers

• Change is scoped to azure-core-amqp; SendLinkHandler is unchanged, so senders that legitimately consume credits are unaffected.
• CHANGELOG entry added under 2.13.0-beta.1 → Bugs Fixed.

[github-actions]

Thank you for your contribution @ksalazar-91! We will review the pull request and get back to you soon.

[Copilot]

Pull request overview

This pull request fixes an unbounded memory leak in azure-core-amqp where RequestResponseChannel failed to subscribe to (and therefore drain) SendLinkHandler’s link-credit flux, causing credits to buffer indefinitely for long-lived cached channels.

Changes:

• Subscribe to sendLinkHandler.getLinkCredits() in RequestResponseChannel and register the resulting Disposable for proper cleanup on channel termination.
• Add a regression test asserting the credit flux is subscribed to and that the subscription is disposed on channel close.
• Document the fix in the azure-core-amqp changelog under 2.13.0-beta.1.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.

File	Description
sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RequestResponseChannel.java	Drains send-link credit emissions by subscribing and disposing via the channel’s subscriptions composite.
sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelTest.java	Adds a regression test ensuring the credit flux is subscribed and cancelled on close.
sdk/core/azure-core-amqp/CHANGELOG.md	Adds a Bugs Fixed entry describing the leak and the mitigation.

[conniey]

Because sendLinkHandler.getLinkCredits() is never subscribed to, is a FAIL_OVERFLOW error eventually reported? Sinks.many().unicast().onBackpressureBuffer() is not unbounded iirc. SinksSpecs.java#L114 (The queue is by default 256.)