From 503d84c6b202841a6acf9be4e68cb2a8bcd70d56 Mon Sep 17 00:00:00 2001 From: Arnab Nandy Date: Thu, 16 Jul 2026 03:48:09 +0530 Subject: [PATCH 1/5] Fix AMQP remote connection close cleanup Signed-off-by: Arnab Nandy --- sdk/core/azure-core-amqp/CHANGELOG.md | 3 ++ .../handler/ConnectionHandler.java | 7 ++- .../handler/ConnectionHandlerTest.java | 43 +++++++++++++++++++ 3 files changed, 51 insertions(+), 2 deletions(-) diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index 990eaf0caef5..84d9c901e37e 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/sdk/core/azure-core-amqp/CHANGELOG.md @@ -8,6 +8,9 @@ ### Bugs Fixed +- Fixed connection and transport resources not being disposed when a remote endpoint closes a connection with an AMQP + error condition. ([#49440](https://github.com/Azure/azure-sdk-for-java/issues/49440)) + ### Other Changes ## 2.12.0 (2026-06-08) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java index 40277324e19f..adaff894a5f4 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java @@ -357,10 +357,13 @@ public void onConnectionRemoteClose(Event event) { final ErrorCondition error = connection.getRemoteCondition(); logErrorCondition("onConnectionRemoteClose", connection, error); - if (error == null) { + if (error == null || error.getCondition() == null) { onNext(connection.getRemoteState()); } else { - notifyErrorContext(connection, error); + final Throwable exception + = ExceptionUtil.toException(error.getCondition().toString(), error.getDescription(), getErrorContext()); + + onError(exception); } } diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java index 328b83acc671..bfb3ce26cc2b 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java @@ -6,6 +6,8 @@ import com.azure.core.amqp.AmqpRetryOptions; import com.azure.core.amqp.AmqpTransportType; import com.azure.core.amqp.ProxyOptions; +import com.azure.core.amqp.exception.AmqpErrorCondition; +import com.azure.core.amqp.exception.AmqpException; import com.azure.core.amqp.implementation.AmqpErrorCode; import com.azure.core.amqp.implementation.AmqpMetricsProvider; import com.azure.core.amqp.implementation.ClientConstants; @@ -241,6 +243,47 @@ void onConnectionInit() { assertTrue(actualProperties.isEmpty()); } + @Test + void onConnectionRemoteCloseWithoutErrorSignalsClosed() { + // Arrange + final Connection connection = mock(Connection.class); + when(connection.getRemoteState()).thenReturn(EndpointState.CLOSED); + + final Event event = mock(Event.class); + when(event.getConnection()).thenReturn(connection); + + // Act & Assert + StepVerifier.create(handler.getEndpointStates()) + .expectNext(EndpointState.UNINITIALIZED) + .then(() -> handler.onConnectionRemoteClose(event)) + .expectNext(EndpointState.CLOSED) + .then(handler::close) + .verifyComplete(); + } + + @Test + void onConnectionRemoteCloseWithErrorSignalsError() { + // Arrange + final ErrorCondition errorCondition + = new ErrorCondition(AmqpErrorCode.CONNECTION_FORCED, "Connection idle timeout."); + final Connection connection = mock(Connection.class); + when(connection.getRemoteState()).thenReturn(EndpointState.CLOSED); + when(connection.getRemoteCondition()).thenReturn(errorCondition); + + final Event event = mock(Event.class); + when(event.getConnection()).thenReturn(connection); + + // Act & Assert + StepVerifier.create(handler.getEndpointStates()) + .expectNext(EndpointState.UNINITIALIZED) + .then(() -> handler.onConnectionRemoteClose(event)) + .expectErrorSatisfies(error -> { + assertTrue(error instanceof AmqpException); + assertEquals(AmqpErrorCondition.CONNECTION_FORCED, ((AmqpException) error).getErrorCondition()); + }) + .verify(); + } + @Test void onConnectionCloseMetrics() { // Arrange From b7de016754c69689848c09c7d1c93c7fedaa149b Mon Sep 17 00:00:00 2001 From: Arnab Nandy Date: Thu, 16 Jul 2026 04:04:59 +0530 Subject: [PATCH 2/5] Refactor AMQP error conversion Signed-off-by: Arnab Nandy --- .../implementation/handler/ConnectionHandler.java | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java index adaff894a5f4..625321595a5a 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java @@ -360,10 +360,7 @@ public void onConnectionRemoteClose(Event event) { if (error == null || error.getCondition() == null) { onNext(connection.getRemoteState()); } else { - final Throwable exception - = ExceptionUtil.toException(error.getCondition().toString(), error.getDescription(), getErrorContext()); - - onError(exception); + onError(toException(error)); } } @@ -400,10 +397,12 @@ private void notifyErrorContext(Connection connection, ErrorCondition condition) } // if the remote-peer abruptly closes the connection without issuing close frame issue one - final Throwable exception = ExceptionUtil.toException(condition.getCondition().toString(), - condition.getDescription(), getErrorContext()); + onError(toException(condition)); + } - onError(exception); + private Throwable toException(ErrorCondition condition) { + return ExceptionUtil.toException(condition.getCondition().toString(), condition.getDescription(), + getErrorContext()); } private void logErrorCondition(String eventName, Connection connection, ErrorCondition error) { From e50acfd9717e60d39987a1537cadfe78d54d03ba Mon Sep 17 00:00:00 2001 From: Arnab Nandy Date: Thu, 16 Jul 2026 04:39:16 +0530 Subject: [PATCH 3/5] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../core/amqp/implementation/handler/ConnectionHandler.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java index 625321595a5a..46e4f2e3a6aa 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java @@ -396,7 +396,7 @@ private void notifyErrorContext(Connection connection, ErrorCondition condition) .logExceptionAsError(new IllegalStateException("notifyErrorContext does not have an ErrorCondition.")); } - // if the remote-peer abruptly closes the connection without issuing close frame issue one + // If the remote peer closes with an error condition, propagate it to subscribers. onError(toException(condition)); } From a6af9b33ab6c2f5f342b369ef1fcc9155652e7aa Mon Sep 17 00:00:00 2001 From: Arnab Nandy Date: Thu, 16 Jul 2026 17:32:07 +0530 Subject: [PATCH 4/5] Handle missing AMQP error conditions Signed-off-by: Arnab Nandy --- .../handler/ConnectionHandler.java | 7 +++--- .../handler/ConnectionHandlerTest.java | 25 +++++++++++++++++++ 2 files changed, 29 insertions(+), 3 deletions(-) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java index 46e4f2e3a6aa..11dd23970776 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java @@ -391,9 +391,10 @@ private void notifyErrorContext(Connection connection, ErrorCondition condition) return; } - if (condition == null) { - throw logger - .logExceptionAsError(new IllegalStateException("notifyErrorContext does not have an ErrorCondition.")); + if (condition == null || condition.getCondition() == null) { + onError(logger + .logExceptionAsError(new IllegalStateException("notifyErrorContext does not have an ErrorCondition."))); + return; } // If the remote peer closes with an error condition, propagate it to subscribers. diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java index bfb3ce26cc2b..200c32acd009 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java @@ -27,6 +27,7 @@ import org.apache.qpid.proton.engine.Event; import org.apache.qpid.proton.engine.SslDomain; import org.apache.qpid.proton.engine.SslPeerDetails; +import org.apache.qpid.proton.engine.Transport; import org.apache.qpid.proton.engine.impl.TransportInternal; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -284,6 +285,30 @@ void onConnectionRemoteCloseWithErrorSignalsError() { .verify(); } + @Test + void onTransportClosedWithoutErrorConditionSignalsClearError() { + // Arrange + final Connection connection = mock(Connection.class); + when(connection.getRemoteState()).thenReturn(EndpointState.ACTIVE); + + final Transport transport = mock(Transport.class); + when(transport.getCondition()).thenReturn(new ErrorCondition(null, "")); + + final Event event = mock(Event.class); + when(event.getConnection()).thenReturn(connection); + when(event.getTransport()).thenReturn(transport); + + // Act & Assert + StepVerifier.create(handler.getEndpointStates()) + .expectNext(EndpointState.UNINITIALIZED) + .then(() -> handler.onTransportClosed(event)) + .expectErrorSatisfies(error -> { + assertTrue(error instanceof IllegalStateException); + assertEquals("notifyErrorContext does not have an ErrorCondition.", error.getMessage()); + }) + .verify(); + } + @Test void onConnectionCloseMetrics() { // Arrange From c6b00b92138135503cb5522620ac2505c2e5a39d Mon Sep 17 00:00:00 2001 From: Arnab Nandy Date: Thu, 16 Jul 2026 17:42:11 +0530 Subject: [PATCH 5/5] Clarify missing AMQP condition diagnostics Signed-off-by: Arnab Nandy --- .../amqp/implementation/handler/ConnectionHandler.java | 8 +++++++- .../implementation/handler/ConnectionHandlerTest.java | 2 +- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java index 11dd23970776..0c6b4e6ce46c 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java @@ -391,12 +391,18 @@ private void notifyErrorContext(Connection connection, ErrorCondition condition) return; } - if (condition == null || condition.getCondition() == null) { + if (condition == null) { onError(logger .logExceptionAsError(new IllegalStateException("notifyErrorContext does not have an ErrorCondition."))); return; } + if (condition.getCondition() == null) { + onError(logger.logExceptionAsError( + new IllegalStateException("notifyErrorContext ErrorCondition does not have a condition symbol."))); + return; + } + // If the remote peer closes with an error condition, propagate it to subscribers. onError(toException(condition)); } diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java index 200c32acd009..932a347163fe 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/ConnectionHandlerTest.java @@ -304,7 +304,7 @@ void onTransportClosedWithoutErrorConditionSignalsClearError() { .then(() -> handler.onTransportClosed(event)) .expectErrorSatisfies(error -> { assertTrue(error instanceof IllegalStateException); - assertEquals("notifyErrorContext does not have an ErrorCondition.", error.getMessage()); + assertEquals("notifyErrorContext ErrorCondition does not have a condition symbol.", error.getMessage()); }) .verify(); }