Messaging: handle proxy connection close header (Azure#45698)

* use unreleased core-amqp version

* adding support for respond to proxy challenge in new connection

* synching the qpid, core-amqp versions

* update the changelog

* updating qpid-extensions version in bannedDependencies

* sync qpid-extensions, core-amqp versions in more places

* changing the tests to accomodate the changes

* fix checkstyle

Author: anuchandy

eng/versioning/external_dependencies.txt

@@ -32,7 +32,7 @@ com.microsoft.azure:azure-client-authentication;1.7.14
 com.microsoft.azure:azure-client-runtime;1.7.14
 com.microsoft.azure:azure-core;0.9.8
 com.microsoft.azure:azure-keyvault-cryptography;1.2.2
-com.microsoft.azure:qpid-proton-j-extensions;1.2.5
+com.microsoft.azure:qpid-proton-j-extensions;1.2.6
 com.microsoft.sqlserver:mssql-jdbc;10.2.3.jre8
 com.microsoft.azure:azure-functions-maven-plugin;1.30.0
 com.microsoft.azure.functions:azure-functions-java-library;2.2.0

eng/versioning/version_client.txt

@@ -530,6 +530,7 @@ io.clientcore:annotation-processor-test;1.0.0-beta.1;1.0.0-beta.1
 # In the pom, the version update tag after the version should name the unreleased package and the dependency version:
 # <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
 
+unreleased_com.azure:azure-core-amqp;2.11.0-beta.1
 unreleased_com.azure.v2:azure-core;2.0.0-beta.1
 unreleased_com.azure.v2:azure-identity;2.0.0-beta.1
 unreleased_io.clientcore:annotation-processor;1.0.0-beta.4

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -4,12 +4,18 @@
 
 ### Features Added
 
+- Added functionality to resume proxy handshake on a new connection if the challenge response closes the existing connection. ([45698](https://github.com/Azure/azure-sdk-for-java/pull/45698)) ([69](https://github.com/Azure/qpid-proton-j-extensions/pull/69))
+
 ### Breaking Changes
 
 ### Bugs Fixed
 
 ### Other Changes
 
+#### Dependency Updates
+
+- Upgraded `qpid-proton-j-extensions` from `1.2.5` to `1.2.6`.
+
 ## 2.10.1 (2025-06-26)
 
 ### Other Changes

sdk/core/azure-core-amqp/pom.xml

@@ -82,7 +82,7 @@
     <dependency>
       <groupId>com.microsoft.azure</groupId>
       <artifactId>qpid-proton-j-extensions</artifactId>
-      <version>1.2.5</version> <!-- {x-version-update;com.microsoft.azure:qpid-proton-j-extensions;external_dependency} -->
+      <version>1.2.6</version> <!-- {x-version-update;com.microsoft.azure:qpid-proton-j-extensions;external_dependency} -->
     </dependency>
     <dependency>
       <groupId>org.apache.qpid</groupId>
@@ -128,7 +128,7 @@
           <rules>
             <bannedDependencies>
               <includes>
-                <include>com.microsoft.azure:qpid-proton-j-extensions:[1.2.5]</include> <!-- {x-include-update;com.microsoft.azure:qpid-proton-j-extensions;external_dependency} -->
+                <include>com.microsoft.azure:qpid-proton-j-extensions:[1.2.6]</include> <!-- {x-include-update;com.microsoft.azure:qpid-proton-j-extensions;external_dependency} -->
                 <include>org.apache.qpid:proton-j:[0.34.1]</include> <!-- {x-include-update;org.apache.qpid:proton-j;external_dependency} -->
               </includes>
             </bannedDependencies>

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java

@@ -215,6 +215,13 @@ public Mono<ReactorConnection> connectAndAwaitToActive() {
         }).thenReturn(this);
     }
 
+    public void transferState(ReactorConnection fromConnection) {
+        if (fromConnection == null) {
+            return;
+        }
+        this.handler.transferState(fromConnection.handler);
+    }
+
     /**
      * {@inheritDoc}
      */

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnectionCache.java

@@ -18,6 +18,7 @@
 import java.util.Objects;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
 import static com.azure.core.amqp.implementation.ClientConstants.CONNECTION_ID_KEY;
@@ -47,6 +48,7 @@ public final class ReactorConnectionCache<T extends ReactorConnection> implement
     // any dependent type; instead, the dependent type must acquire Connection only through the cache route,
     // i.e., by subscribing to 'createOrGetCachedConnection' via 'get()' getter.
     private volatile T currentConnection;
+    private final State state = new State();
 
     /**
      * Create a ReactorConnectionCache that is responsible for obtaining a connection, waiting for it to active,
@@ -86,6 +88,7 @@ public ReactorConnectionCache(Supplier<T> connectionSupplier, String fullyQualif
         });
 
         this.createOrGetCachedConnection = newConnection.flatMap(c -> {
+            state.transfer(c);
             withConnectionId(logger, c.getId()).log("Waiting to connect and active.");
 
             return c.connectAndAwaitToActive().doOnCancel(() -> {
@@ -289,4 +292,15 @@ private static void closeConnection(ReactorConnection c, ClientLogger logger, St
     private static LoggingEventBuilder withConnectionId(ClientLogger logger, String id) {
         return logger.atInfo().addKeyValue(CONNECTION_ID_KEY, id);
     }
+
+    private static final class State {
+        private final AtomicReference<ReactorConnection> s = new AtomicReference<>(null);
+
+        void transfer(ReactorConnection c) {
+            final ReactorConnection from = s.getAndSet(c);
+            if (from != null) {
+                c.transferState(from);
+            }
+        }
+    }
 }

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/ConnectionHandler.java

@@ -131,6 +131,10 @@ public ConnectionHandler(final String connectionId, final ConnectionOptions conn
         this.enableSsl = enableSsl;
     }
 
+    public void transferState(ConnectionHandler fromHandler) {
+        // No state to transfer in ConnectionHandler
+    }
+
     /**
      * Gets properties to add when creating AMQP connection.
      *

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/WebSocketsProxyConnectionHandler.java

@@ -56,6 +56,7 @@ public class WebSocketsProxyConnectionHandler extends WebSocketsConnectionHandle
      * or Service Bus.
      */
     private final String connectHostnameAndPort;
+    private final ProxyImpl proxy;
 
     /**
      * Creates a handler that handles proton-j's connection through a proxy using web sockets.
@@ -97,6 +98,12 @@ public WebSocketsProxyConnectionHandler(String connectionId, ConnectionOptions c
             final Proxy proxy = proxies.get(0);
             this.proxyHostAddress = (InetSocketAddress) proxy.address();
         }
+
+        if (proxyOptions == ProxyOptions.SYSTEM_DEFAULTS) {
+            this.proxy = new ProxyImpl();
+        } else {
+            this.proxy = new ProxyImpl(getProtonConfiguration());
+        }
     }
 
     /**
@@ -120,6 +127,16 @@ public static boolean shouldUseProxy(final String hostname, final int port) {
         return isProxyAddressLegal(proxies);
     }
 
+    @Override
+    public void transferState(ConnectionHandler fromHandler) {
+        if (fromHandler instanceof WebSocketsProxyConnectionHandler) {
+            final WebSocketsProxyConnectionHandler wsHandler = (WebSocketsProxyConnectionHandler) fromHandler;
+            final ProxyImpl fromProxy = wsHandler.proxy;
+            this.proxy.transferState(fromProxy);
+        }
+        super.transferState(fromHandler);
+    }
+
     /**
      * Gets the hostname for the proxy.
      *
@@ -205,11 +222,6 @@ public void onTransportError(Event event) {
     protected void addTransportLayers(final Event event, final TransportInternal transport) {
         super.addTransportLayers(event, transport);
 
-        // Checking that the proxy configuration is not null and not equal to the system defaults option.
-        final ProxyImpl proxy = proxyOptions != null && !(proxyOptions == ProxyOptions.SYSTEM_DEFAULTS)
-            ? new ProxyImpl(getProtonConfiguration())
-            : new ProxyImpl();
-
         final ProxyHandler proxyHandler = new ProxyHandlerImpl();
         proxy.configure(connectHostnameAndPort, null, proxyHandler, transport);
 

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/WebSocketsProxyConnectionHandlerTest.java

@@ -201,11 +201,10 @@ public void proxyConfigureConnectHostnameAndPortDerivesFromFqdn() {
         final Proxy newProxy = new Proxy(Proxy.Type.HTTP, address);
         final ProxyOptions proxyOptions = new ProxyOptions(ProxyAuthenticationType.BASIC, newProxy, USERNAME, PASSWORD);
 
-        this.handler = new WebSocketsProxyConnectionHandler(CONNECTION_ID, connectionOptions, proxyOptions, peerDetails,
-            AmqpMetricsProvider.noop());
-
         // Act and Assert
         try (MockedConstruction<ProxyImpl> mockConstruction = mockConstruction(ProxyImpl.class)) {
+            this.handler = new WebSocketsProxyConnectionHandler(CONNECTION_ID, connectionOptions, proxyOptions,
+                peerDetails, AmqpMetricsProvider.noop());
             this.handler.addTransportLayers(mock(Event.class, Mockito.CALLS_REAL_METHODS),
                 mock(TransportImpl.class, Mockito.CALLS_REAL_METHODS));
 
@@ -236,11 +235,10 @@ public void proxyConfigureConnectHostnameAndPortDerivesFromCustomEndpoint() {
                 AmqpTransportType.AMQP_WEB_SOCKETS, new AmqpRetryOptions(), ProxyOptions.SYSTEM_DEFAULTS, scheduler,
                 CLIENT_OPTIONS, VERIFY_MODE, PRODUCT, CLIENT_VERSION, customEndpointHostname, customEndpointPort, true);
 
-        this.handler = new WebSocketsProxyConnectionHandler(CONNECTION_ID, connectionOptionsWithCustomEndpoint,
-            proxyOptions, peerDetails, AmqpMetricsProvider.noop());
-
         // Act and Assert
         try (MockedConstruction<ProxyImpl> mockConstruction = mockConstruction(ProxyImpl.class)) {
+            this.handler = new WebSocketsProxyConnectionHandler(CONNECTION_ID, connectionOptionsWithCustomEndpoint,
+                proxyOptions, peerDetails, AmqpMetricsProvider.noop());
             this.handler.addTransportLayers(mock(Event.class, Mockito.CALLS_REAL_METHODS),
                 mock(TransportImpl.class, Mockito.CALLS_REAL_METHODS));
 

sdk/eventhubs/azure-messaging-eventhubs/pom.xml

@@ -53,7 +53,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core-amqp</artifactId>
-      <version>2.10.1</version> <!-- {x-version-update;com.azure:azure-core-amqp;dependency} -->
+      <version>2.11.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core-amqp;dependency} -->
     </dependency>
 
     <!-- Test dependencies -->