From 9e8855a6f5288c043f430b8070afc0ec857ccf77 Mon Sep 17 00:00:00 2001
From: Suresh Kumar Anaparti <sureshkumar.anaparti@gmail.com>
Date: Mon, 6 Jul 2026 16:24:34 +0530
Subject: [PATCH] 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>
---
 .../jobs/impl/AsyncJobManagerImpl.java        | 24 +++++++++++++++++--
 1 file changed, 22 insertions(+), 2 deletions(-)

diff --git a/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java b/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
index 7672b9dc6f9..9d71c6d07b4 100644
--- a/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
+++ b/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 class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
     @Override
     public boolean stop() {
         _heartbeatScheduler.shutdown();
+        _eventBusPublisher.shutdown();
         _apiJobExecutor.shutdown();
         _workerJobExecutor.shutdown();
         return true;
@@ -1397,8 +1399,26 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
     }
 
     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