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..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 @@ -357,10 +357,10 @@ 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); + onError(toException(error)); } } @@ -392,15 +392,24 @@ private void notifyErrorContext(Connection connection, ErrorCondition condition) } if (condition == null) { - throw logger - .logExceptionAsError(new IllegalStateException("notifyErrorContext does not have an ErrorCondition.")); + 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 abruptly closes the connection without issuing close frame issue one - final Throwable exception = ExceptionUtil.toException(condition.getCondition().toString(), - condition.getDescription(), getErrorContext()); + // If the remote peer closes with an error condition, propagate it to subscribers. + 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) { 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..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 @@ -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; @@ -25,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; @@ -241,6 +244,71 @@ 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 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 ErrorCondition does not have a condition symbol.", error.getMessage()); + }) + .verify(); + } + @Test void onConnectionCloseMetrics() { // Arrange