Add async job event bus publisher (#12971)

* Async event bus publishing in AsyncJobManagerImpl to reduce API thread contention

publishOnEventBus() was calling _messageBus.publish() synchronously on the
request thread, which blocks on MessageBusBase$Gate (an exclusive mutex).
JFR analysis showed this causing up to 107ms waits on Jetty request threads,
contributing to 502 errors from the upstream load balancer.

Move event bus publishing to a dedicated single-threaded executor so API
request threads are no longer blocked by Gate contention. Event ordering
is preserved by the single-threaded executor.

* Wrap async event bus publish in ManagedContextRunnable for DB connection safety

The sole subscriber (ApiServer.handleAsyncJobPublishEvent) performs DAO reads
(getUserIncludingRemoved, getAccount, findById) inside its callback.
Without ManagedContextRunnable, the EventBus thread would not have proper
TransactionLegacy lifecycle management, risking DB connection leaks.

* logger fix

* review comments

* review changes

---------

Co-authored-by: Aaron Chung <aaron_chung@apple.com>

Author: sureshanaparti

framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java

@@ -184,6 +184,7 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
     private volatile long _executionRunNumber = 1;
 
     private final ScheduledExecutorService _heartbeatScheduler = Executors.newScheduledThreadPool(1, new NamedThreadFactory("AsyncJobMgr-Heartbeat"));
+    private final ExecutorService _eventBusPublisher = Executors.newSingleThreadExecutor(new NamedThreadFactory("AsyncJobMgr-EventBus"));
     private ExecutorService _apiJobExecutor;
     private ExecutorService _workerJobExecutor;
 
@@ -1378,6 +1379,7 @@ public boolean start() {
     @Override
     public boolean stop() {
         _heartbeatScheduler.shutdown();
+        _eventBusPublisher.shutdown();
         _apiJobExecutor.shutdown();
         _workerJobExecutor.shutdown();
         return true;
@@ -1397,8 +1399,26 @@ protected AsyncJobManagerImpl() {
     }
 
     private void publishOnEventBus(AsyncJob job, String jobEvent) {
-        _messageBus.publish(null, AsyncJob.Topics.JOB_EVENT_PUBLISH, PublishScope.LOCAL,
-            new Pair<AsyncJob, String>(job, jobEvent));
+        try {
+            _eventBusPublisher.submit(new ManagedContextRunnable() {
+                @Override
+                protected void runInContext() {
+                    publishJobEvent(job, jobEvent);
+                }
+            });
+        } catch (RejectedExecutionException e) {
+            logger.warn("Failed to publish async job event, event bus publisher is shut down", e);
+        }
+    }
+
+    private void publishJobEvent(AsyncJob job, String jobEvent) {
+        try {
+            _messageBus.publish(null, AsyncJob.Topics.JOB_EVENT_PUBLISH, PublishScope.LOCAL,
+                    new Pair<>(job, jobEvent));
+        } catch (Throwable t) {
+            logger.warn("Failed to publish async job event on message bus. jobId={}, jobEvent={}",
+                    job != null ? job.getId() : null, jobEvent, t);
+        }
     }
 
     @Override