Fix AMQP remote connection close cleanup

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)

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) {

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