Skip to content

Add async job event bus publisher #12971

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sureshanaparti wants to merge 5 commits into
base: 4.22
Choose a base branch
from
+22 −2
Open

Add async job event bus publisher #12971

Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
68c69df
Async event bus publishing in AsyncJobManagerImpl to reduce API threa…
sliceofapplepie Mar 10, 2026
26e5d3c
Wrap async event bus publish in ManagedContextRunnable for DB connect…
sliceofapplepie Mar 10, 2026
8f996ac
logger fix
sureshanaparti Apr 7, 2026
97b15fc
review comments
sureshanaparti Apr 7, 2026
1848fb1
review changes
sureshanaparti Apr 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions ...ork/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Comment on lines 186 to 188
Copy link

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Executors.newSingleThreadExecutor(...) uses an unbounded LinkedBlockingQueue. If _messageBus.publish() blocks (the original issue) for extended periods, this queue can grow without bound and increase memory pressure or OOM under sustained load. Consider using a bounded ThreadPoolExecutor (still single-threaded) with an explicit queue size and a rejection policy (e.g., log+drop or CallerRuns as a backpressure mechanism) that matches desired guarantees for job event delivery.

Copilot uses AI. Check for mistakes.
private ExecutorService _workerJobExecutor;

Expand Down Expand Up @@ -1378,6 +1379,7 @@ public boolean start() {
@Override
public boolean stop() {
_heartbeatScheduler.shutdown();
_eventBusPublisher.shutdown();
_apiJobExecutor.shutdown();
_workerJobExecutor.shutdown();
Comment on lines 1379 to 1384
Copy link

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

stop() only calls _eventBusPublisher.shutdown() and immediately returns. Because NamedThreadFactory creates non-daemon threads, a slow/stuck publish can keep the JVM alive longer than expected during shutdown, and queued events may continue to run while other components are already stopping. Consider awaiting termination with a timeout and then shutdownNow() (with logging) to ensure predictable shutdown behavior.

Copilot uses AI. Check for mistakes.
return true;
Expand All @@ -1397,8 +1399,17 @@ 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() {
_messageBus.publish(null, AsyncJob.Topics.JOB_EVENT_PUBLISH, PublishScope.LOCAL,
new Pair<AsyncJob, String>(job, jobEvent));
Copy link

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Exceptions thrown by _messageBus.publish(...) inside the submitted task will be captured by the executor and effectively ignored because the returned Future is not observed. This can silently drop job events while the API call continues normally. Consider wrapping the publish call in a try/catch (at least Exception, ideally Throwable) within runInContext() and logging failures (optionally including jobId/jobEvent) so operational issues are visible.

Suggested change
_messageBus.publish(null, AsyncJob.Topics.JOB_EVENT_PUBLISH, PublishScope.LOCAL,
new Pair<AsyncJob, String>(job, jobEvent));
try {
_messageBus.publish(null, AsyncJob.Topics.JOB_EVENT_PUBLISH, PublishScope.LOCAL,
new Pair<AsyncJob, String>(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);
}

Copilot uses AI. Check for mistakes.
}
});
Comment on lines 1401 to +1408
Copy link

Copilot AI Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

publishOnEventBus now captures and publishes the live AsyncJob object asynchronously. Callers frequently mutate the same AsyncJobVO instance after calling publishOnEventBus (e.g., updateAsyncJobStatus mutates job inside the following transaction), so subscribers may observe a job state that no longer corresponds to the jobEvent being emitted (notably the "submit" event can be delivered with a job already in-progress or completed). To preserve event semantics, publish an immutable snapshot (e.g., jobId + jobEvent + selected fields copied at enqueue time, or re-fetch a fresh immutable view inside the publisher thread).

Copilot uses AI. Check for mistakes.
} catch (RejectedExecutionException e) {
logger.warn("Failed to publish async job event, event bus publisher is shut down", e);
}
Comment on lines +1402 to +1411
Copy link
Copy Markdown
Contributor

@DaanHoogland DaanHoogland Apr 9, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nested try catch block, can you make a new method for the inner one?

Copy link
Copy Markdown
Contributor Author

@sureshanaparti sureshanaparti Apr 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

}

@Override
Expand Down
Loading