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

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -8,6 +8,12 @@
 
 ### Bugs Fixed
 
+- Fixed a memory leak in `RequestResponseChannel` (used by management/request-response operations such as
+  lock renewal, peek, schedule, and session state). The channel's `SendLinkHandler` emits a link-credit value
+  on every AMQP flow frame into a unicast, unbounded-buffer sink, but the channel never subscribed to it, so the
+  credits buffered indefinitely and the heap grew steadily for long-lived, cached channels. The channel now drains
+  the credit flux. ([#47261](https://github.com/Azure/azure-sdk-for-java/issues/47261))
+
 ### Other Changes
 
 ## 2.12.0 (2026-06-08)

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RequestResponseChannel.java

@@ -221,6 +221,15 @@ protected RequestResponseChannel(AmqpConnection amqpConnection, String connectio
         });
         this.subscriptions.add(sendEndpointDisposable);
 
+        // Drain the send link's credit flux. Unlike ReactorSender, RequestResponseChannel does not use AMQP
+        // link-credit based flow control (it correlates responses via 'unconfirmedSends'), so it never subscribed
+        // to the credits. However, SendLinkHandler.getLinkCredits() is backed by a unicast onBackpressureBuffer sink
+        // that emits a value on every AMQP flow frame (onLinkFlow); with no subscriber those values buffer forever,
+        // leaking memory for long-lived, cached management channels. Subscribe and discard to keep the buffer drained.
+        // https://github.com/Azure/azure-sdk-for-java/issues/47261
+        final Disposable sendCreditsDisposable = sendLinkHandler.getLinkCredits().subscribe();
+        this.subscriptions.add(sendCreditsDisposable);
+
         // To ensure graceful closure of request-response-channel instance that won the race between
         // its creation and its parent connection close.
         final Disposable shutdownDisposable = amqpConnection.getShutdownSignals().next().flatMap(signal -> {

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelTest.java

@@ -72,6 +72,7 @@ class RequestResponseChannelTest {
     private final TestPublisher<Delivery> deliveryProcessor = TestPublisher.createCold();
     private final TestPublisher<EndpointState> receiveEndpoints = TestPublisher.createCold();
     private final TestPublisher<EndpointState> sendEndpoints = TestPublisher.createCold();
+    private final TestPublisher<Integer> sendCredits = TestPublisher.createCold();
     private final TestPublisher<AmqpShutdownSignal> shutdownSignals = TestPublisher.create();
     private final boolean isV2 = false;
 
@@ -122,6 +123,7 @@ void beforeEach() throws IOException {
         when(receiveLinkHandler.getEndpointStates()).thenReturn(receiveEndpoints.flux());
         when(receiveLinkHandler.getDeliveredMessages()).thenReturn(deliveryProcessor.flux());
         when(sendLinkHandler.getEndpointStates()).thenReturn(sendEndpoints.flux());
+        when(sendLinkHandler.getLinkCredits()).thenReturn(sendCredits.flux());
 
         when(amqpConnection.getShutdownSignals()).thenReturn(shutdownSignals.flux());
 
@@ -206,6 +208,31 @@ ENTITY_PATH, sessionWrapper(session), retryOptions, handlerProvider, reactorProv
         assertTrue(channel.isDisposed());
     }
 
+    /**
+     * Verifies that the channel subscribes to the send link credit flux so it is drained, preventing the
+     * unbounded credit buffer leak described in https://github.com/Azure/azure-sdk-for-java/issues/47261.
+     */
+    @Test
+    void subscribesToSendLinkCreditsToPreventLeak() {
+        // Arrange & Act
+        final RequestResponseChannel channel = new RequestResponseChannel(amqpConnection, CONNECTION_ID, NAMESPACE,
+            ENTITY_PATH, sessionWrapper(session), retryOptions, handlerProvider, reactorProvider, serializer,
+            SenderSettleMode.SETTLED, ReceiverSettleMode.SECOND, AmqpMetricsProvider.noop(), isV2);
+
+        // Assert: the channel subscribed to the credit flux (so emitted credits are consumed, not buffered forever).
+        sendCredits.assertSubscribers();
+
+        // Closing the channel disposes the credit subscription.
+        receiveEndpoints.next(EndpointState.ACTIVE);
+        sendEndpoints.next(EndpointState.ACTIVE);
+        StepVerifier.create(channel.closeAsync()).then(() -> {
+            sendEndpoints.complete();
+            receiveEndpoints.complete();
+        }).expectComplete().verify(VERIFY_TIMEOUT);
+
+        sendCredits.assertCancelled();
+    }
+
     @Test
     void dispose() throws IOException {
         // Arrange