From aeff6fa903c6e7e786a72bb4b2a28a6cb502560c Mon Sep 17 00:00:00 2001 From: ksalazar-91 Date: Thu, 9 Jul 2026 14:51:30 -0700 Subject: [PATCH 1/2] Fix memory leak in RequestResponseChannel by draining SendLinkHandler link credits (#47261) RequestResponseChannel never subscribed to SendLinkHandler.getLinkCredits(), a unicast unbounded-buffer sink fed on every AMQP flow frame, so credits buffered indefinitely for long-lived cached management channels. The channel now drains the credit flux (disposed on close), mirroring ReactorSender. Adds a regression test and CHANGELOG entry. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: f2d0e805-73ed-4d31-9bb2-b4541c325600 --- sdk/core/azure-core-amqp/CHANGELOG.md | 6 +++++ .../RequestResponseChannel.java | 9 +++++++ .../RequestResponseChannelTest.java | 27 +++++++++++++++++++ 3 files changed, 42 insertions(+) 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 From f6d0922073e78a7db1936a31768b8f7a2fa7ed99 Mon Sep 17 00:00:00 2001 From: ksalazar-91 Date: Wed, 15 Jul 2026 18:36:42 -0700 Subject: [PATCH 2/2] Stub SendLinkHandler.getLinkCredits() in RequestResponseChannelCacheTest (#47261) The credit-drain fix makes RequestResponseChannel subscribe to getLinkCredits() during construction. RequestResponseChannelCacheTest mocks SendLinkHandler but did not stub getLinkCredits(), so it returned null and the new subscribe() threw NullPointerException, failing CI. Stub it with Flux.never() (matching the sibling getMessages() stub). Test-only; production getLinkCredits() never returns null. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: f2d0e805-73ed-4d31-9bb2-b4541c325600 --- .../amqp/implementation/RequestResponseChannelCacheTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelCacheTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelCacheTest.java index 7e6765518424..d4fe45a9f3fd 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelCacheTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/RequestResponseChannelCacheTest.java @@ -459,6 +459,7 @@ void arrange(ReactorConnection connection, String fqdn, String chLinkName, Strin final SendLinkHandler sendLinkHandler = mock(SendLinkHandler.class); when(receiveLinkHandler.getEndpointStates()).thenReturn(receiveLinkStates.asFlux()); when(sendLinkHandler.getEndpointStates()).thenReturn(sendLinkStates.asFlux()); + when(sendLinkHandler.getLinkCredits()).thenReturn(Flux.never()); final ReactorHandlerProvider handlerProvider = mock(ReactorHandlerProvider.class); when(handlerProvider.createReceiveLinkHandler(eq(connectionId), eq(fqdn), eq(chLinkName), eq(chEntityPath), eq(DeliverySettleMode.ACCEPT_AND_SETTLE_ON_DELIVERY), eq(false),