Fix race condition in async exception handling by combining JMS ExceptionListener and TransportListener notifications

Author: jeanouii

activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java

@@ -2046,8 +2046,21 @@ public void onAsyncException(Throwable error) {
 
     @Override
     public void onException(final IOException error) {
-        onAsyncException(error);
+        // Combine JMS ExceptionListener and TransportListener notifications
+        // into a single async task to prevent a race condition where the
+        // ExceptionListener (e.g. ConnectionPool) closes the connection and
+        // shuts down the executor before the TransportListener task is queued.
         executeAsync(() -> {
+            // Notify JMS ExceptionListener first (same as onAsyncException)
+            if (exceptionListener != null) {
+                try {
+                    final JMSException jmsError = JMSExceptionSupport.create(error);
+                    exceptionListener.onException(jmsError);
+                } catch (final Exception e) {
+                    LOG.debug("Exception during JMS ExceptionListener notification", e);
+                }
+            }
+
             transportFailed(error);
             ServiceSupport.dispose(ActiveMQConnection.this.transport);
             brokerInfoReceived.countDown();