Fixes the NullPointerException in RequestResponseChannel when the thread constructing RequestResponseChannel happens to run the cleanup phase synchronously (Azure#36613)

Author: anuchandy

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

@@ -8,6 +8,8 @@
 
 ### Bugs Fixed
 
+- Fixes the NullPointerException in RequestResponseChannel when the thread constructing RequestResponseChannel happens to run the cleanup phase synchronously. ([36607](https://github.com/Azure/azure-sdk-for-java/issues/36607))
+
 ### Other Changes
 
 ## 2.9.0-beta.5 (2023-08-11)

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

@@ -96,7 +96,7 @@ public class RequestResponseChannel implements AsyncCloseable {
     private final AtomicBoolean isDisposed = new AtomicBoolean();
     // Tracks all subscriptions listening for events from various endpoints (sender, receiver & connection),
     // those subscriptions should be disposed when the request-response-channel terminates.
-    private final Disposable.Composite subscriptions;
+    private final Disposable.Composite subscriptions = Disposables.composite();
 
     private final AmqpRetryOptions retryOptions;
     private final String replyTo;
@@ -179,46 +179,46 @@ protected RequestResponseChannel(AmqpConnection amqpConnection, String connectio
 
         // Subscribe to the events from endpoints (Sender, Receiver & Connection) and track the subscriptions.
         //
-        //@formatter:off
-        this.subscriptions = Disposables.composite(
-            receiveLinkHandler.getDeliveredMessages()
-                .map(this::decodeDelivery)
-                .subscribe(message -> {
-                    logger.atVerbose()
-                        .addKeyValue("messageId", message.getCorrelationId())
-                        .log("Settling message.");
-
-                    settleMessage(message);
-                }),
-
-            receiveLinkHandler.getEndpointStates().subscribe(state -> {
-                updateEndpointState(null, AmqpEndpointStateUtil.getConnectionState(state));
-            }, error -> {
-                handleError(error, "Error in ReceiveLinkHandler.");
-                onTerminalState("ReceiveLinkHandler");
-            }, () -> {
-                closeAsync().subscribe();
-                onTerminalState("ReceiveLinkHandler");
-            }),
-
-            sendLinkHandler.getEndpointStates().subscribe(state -> {
-                updateEndpointState(AmqpEndpointStateUtil.getConnectionState(state), null);
-            }, error -> {
-                handleError(error, "Error in SendLinkHandler.");
-                onTerminalState("SendLinkHandler");
-            }, () -> {
-                closeAsync().subscribe();
-                onTerminalState("SendLinkHandler");
-            }),
-
-            // To ensure graceful closure of request-response-channel instance that won the race between
-            // its creation and its parent connection close.
-            amqpConnection.getShutdownSignals().next().flatMap(signal -> {
-                logger.verbose("Shutdown signal received.");
-                return closeAsync();
-            }).subscribe()
-        );
-        //@formatter:on
+        final Disposable receiveMessagesDisposable = receiveLinkHandler.getDeliveredMessages()
+            .map(this::decodeDelivery)
+            .subscribe(message -> {
+                logger.atVerbose()
+                    .addKeyValue("messageId", message.getCorrelationId())
+                    .log("Settling message.");
+
+                settleMessage(message);
+            });
+        this.subscriptions.add(receiveMessagesDisposable);
+
+        final Disposable receiveEndpointDisposable = receiveLinkHandler.getEndpointStates().subscribe(state -> {
+            updateEndpointState(null, AmqpEndpointStateUtil.getConnectionState(state));
+        }, error -> {
+            handleError(error, "Error in ReceiveLinkHandler.");
+            onTerminalState("ReceiveLinkHandler");
+        }, () -> {
+            closeAsync().subscribe();
+            onTerminalState("ReceiveLinkHandler");
+        });
+        this.subscriptions.add(receiveEndpointDisposable);
+
+        final Disposable sendEndpointDisposable = sendLinkHandler.getEndpointStates().subscribe(state -> {
+            updateEndpointState(AmqpEndpointStateUtil.getConnectionState(state), null);
+        }, error -> {
+            handleError(error, "Error in SendLinkHandler.");
+            onTerminalState("SendLinkHandler");
+        }, () -> {
+            closeAsync().subscribe();
+            onTerminalState("SendLinkHandler");
+        });
+        this.subscriptions.add(sendEndpointDisposable);
+
+        // 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 -> {
+            logger.verbose("Shutdown signal received.");
+            return closeAsync();
+        }).subscribe();
+        this.subscriptions.add(shutdownDisposable);
 
         // Open send and receive links.
         //