PROTON-2847 Ensure notification executor is shutdown on close

When connection fails or is closed, ensure the notification executor is shutdown.
apache · Sep 4, 2024 · 6a0c492 · 6a0c492
1 parent b45180b
commit 6a0c492
Showing 1 changed file with 32 additions and 14 deletions.
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java
@@ -559,6 +559,7 @@ ClientConnection connect() throws ClientException {
             openFuture.failed(failureCause);
             closeFuture.complete(this);
             ioContext.shutdown();
+            notifications.shutdown();
 
             throw failureCause;
         }
@@ -777,23 +778,24 @@ private void handleEngineShutdown(Engine engine) {
 
         ioContext.shutdownAsync();
 
-        if (failureCause != null)
-        {
-            openFuture.failed(failureCause);
-            closeFuture.complete(this);
+        try {
+            if (failureCause != null) {
+                openFuture.failed(failureCause);
+                closeFuture.complete(this);
 
-            LOG.warn("Connection {} has failed due to: {}", getId(), failureCause != null ?
-                    failureCause.getClass().getSimpleName() + " -> " + failureCause.getMessage() : "No failure details provided.");
+                LOG.warn("Connection {} has failed due to: {}", getId(), failureCause != null ?
+                         failureCause.getClass().getSimpleName() + " -> " + failureCause.getMessage() : "No failure details provided.");
 
-           submitDisconnectionEvent(options.disconnectedHandler(), transport.getHost(), transport.getPort(), failureCause);
-        }
-        else
-        {
-            openFuture.complete(this);
-            closeFuture.complete(this);
-        }
+               submitDisconnectionEvent(options.disconnectedHandler(), transport.getHost(), transport.getPort(), failureCause);
+            } else {
+                openFuture.complete(this);
+                closeFuture.complete(this);
+            }
 
-        client.unregisterConnection(this);
+            client.unregisterConnection(this);
+        } finally {
+            submitNotificationShutdownTask();
+        }
     }
 
     private void submitConnectionEvent(BiConsumer<Connection, ConnectionEvent> handler, String host, int port, ClientIOException cause) {
@@ -828,6 +830,22 @@ private void submitDisconnectionEvent(BiConsumer<Connection, DisconnectionEvent>
         }
     }
 
+    private void submitNotificationShutdownTask() {
+        try {
+            if (!notifications.isShutdown()) {
+                notifications.submit(() -> {
+                    try {
+                        notifications.shutdown();
+                    } catch (Exception ex) {
+                        LOG.trace("Shutdown of notification event handler threw an error: ", ex);
+                    }
+                });
+            }
+        } catch (Exception ex) {
+            LOG.trace("Error thrown while attempting to submit shutdown of notification executor: ", ex);
+        }
+    }
+
     private Engine configureEngineSaslSupport() {
         if (options.saslOptions().saslEnabled()) {
             SaslMechanismSelector mechSelector =