[Service Bus] Shutdown scheduler before closing client (Azure#17738)

* Shutdown scheduler before closing client

* Change equality check

* Refactored close logic

Author: srnagar

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusProcessorClient.java

@@ -120,16 +120,25 @@ public synchronized void start() {
             logger.info("Processor is already running");
             return;
         }
+
+        if (asyncClient.get() == null) {
+            ServiceBusReceiverAsyncClient newReceiverClient = this.receiverBuilder == null
+                ? this.sessionReceiverBuilder.buildAsyncClientForProcessor()
+                : this.receiverBuilder.buildAsyncClient();
+            asyncClient.set(newReceiverClient);
+        }
+
         receiveMessages();
 
         // Start an executor to periodically check if the client's connection is active
-        this.scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
-        scheduledExecutor.scheduleWithFixedDelay(() -> {
-            if (this.asyncClient.get().isConnectionClosed()) {
-                restartMessageReceiver();
-            }
-        }, SCHEDULER_INTERVAL_IN_SECONDS, SCHEDULER_INTERVAL_IN_SECONDS, TimeUnit.SECONDS);
-
+        if (this.scheduledExecutor == null) {
+            this.scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
+            scheduledExecutor.scheduleWithFixedDelay(() -> {
+                if (this.asyncClient.get().isConnectionClosed()) {
+                    restartMessageReceiver();
+                }
+            }, SCHEDULER_INTERVAL_IN_SECONDS, SCHEDULER_INTERVAL_IN_SECONDS, TimeUnit.SECONDS);
+        }
     }
 
     /**
@@ -149,9 +158,16 @@ public synchronized void close() {
         isRunning.set(false);
         if (receiverSubscription.get() != null) {
             receiverSubscription.get().cancel();
+            receiverSubscription.set(null);
+        }
+        if (scheduledExecutor != null) {
+            scheduledExecutor.shutdown();
+            scheduledExecutor = null;
+        }
+        if (asyncClient.get() != null) {
+            asyncClient.get().close();
+            asyncClient.set(null);
         }
-        asyncClient.get().close();
-        scheduledExecutor.shutdown();
     }
 
     /**
@@ -301,6 +317,9 @@ private void handleError(Throwable throwable) {
     }
 
     private void restartMessageReceiver() {
+        if (!isRunning()) {
+            return;
+        }
         receiverSubscription.set(null);
         ServiceBusReceiverAsyncClient receiverClient = asyncClient.get();
         receiverClient.close();