diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index 990eaf0caef5..b4da8446ebd4 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/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) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RequestResponseChannel.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RequestResponseChannel.java index 3cefba230f0d..9f556905658e 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RequestResponseChannel.java +++ b/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 -> { diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelTest.java index 5f49233de21a..377db8b51d07 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelTest.java @@ -72,6 +72,7 @@ class RequestResponseChannelTest { private final TestPublisher deliveryProcessor = TestPublisher.createCold(); private final TestPublisher receiveEndpoints = TestPublisher.createCold(); private final TestPublisher sendEndpoints = TestPublisher.createCold(); + private final TestPublisher sendCredits = TestPublisher.createCold(); private final TestPublisher 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