(cloudstack) branch 4.22 updated: Add async job event bus publisher (#12971)

[nv...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

nvazquez pushed a commit to branch 4.22
in repository https://gitbox.apache.org/repos/asf/cloudstack.git


The following commit(s) were added to refs/heads/4.22 by this push:
     new 9e8855a6f52 Add async job event bus publisher (#12971)
9e8855a6f52 is described below

commit 9e8855a6f5288c043f430b8070afc0ec857ccf77
Author: Suresh Kumar Anaparti <su...@gmail.com>
AuthorDate: Mon Jul 6 16:24:34 2026 +0530

    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 <aa...@apple.com>
---
 .../framework/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