Use own MainEventBus for each test rather than a shared instance

Author: kabir

.github/workflows/build-and-test.yml

@@ -26,7 +26,7 @@ jobs:
         distribution: 'temurin'
         cache: maven
     - name: Build with Maven and run tests
-      run: mvn -B package --file pom.xml -fae
+      run: mvn -B install --file pom.xml -fae
     - name: Upload Test Reports
       if: failure()
       uses: actions/upload-artifact@v4

extras/queue-manager-replicated/core/src/test/java/io/a2a/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java

@@ -357,7 +357,10 @@ void testReplicatedEventProcessedWhenTaskActive() throws InterruptedException {
         EventQueue childQueue = queue.tap();
         Event dequeuedEvent;
         try {
-            dequeuedEvent = childQueue.dequeueEventItem(100).getEvent();
+            // Wait up to 5 seconds for event (async processing needs time)
+            EventQueueItem item = childQueue.dequeueEventItem(5000);
+            assertNotNull(item, "Event should be available in queue");
+            dequeuedEvent = item.getEvent();
         } catch (EventQueueClosedException e) {
             fail("Queue should not be closed");
             return;

server-common/src/test/java/io/a2a/server/events/EventConsumerTest.java

@@ -15,6 +15,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+import io.a2a.server.tasks.InMemoryTaskStore;
+import io.a2a.server.tasks.PushNotificationSender;
 import io.a2a.spec.A2AError;
 import io.a2a.spec.A2AServerException;
 import io.a2a.spec.Artifact;
@@ -28,14 +30,18 @@
 import io.a2a.spec.TaskStatusUpdateEvent;
 import io.a2a.spec.TextPart;
 import io.a2a.util.Utils;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 public class EventConsumerTest {
 
+    private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = task -> {};
+
     private EventQueue eventQueue;
     private EventConsumer eventConsumer;
-
+    private MainEventBus mainEventBus;
+    private MainEventBusProcessor mainEventBusProcessor;
 
     private static final String MINIMAL_TASK = """
             {
@@ -57,10 +63,25 @@ public class EventConsumerTest {
 
     @BeforeEach
     public void init() {
-        eventQueue = EventQueueUtil.getEventQueueBuilder().build().tap();
+        // Set up MainEventBus and processor for production-like test environment
+        InMemoryTaskStore taskStore = new InMemoryTaskStore();
+        mainEventBus = new MainEventBus();
+        mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER);
+        EventQueueUtil.start(mainEventBusProcessor);
+
+        eventQueue = EventQueueUtil.getEventQueueBuilder()
+                .mainEventBus(mainEventBus)
+                .build().tap();
         eventConsumer = new EventConsumer(eventQueue);
     }
 
+    @AfterEach
+    public void cleanup() {
+        if (mainEventBusProcessor != null) {
+            EventQueueUtil.stop(mainEventBusProcessor);
+        }
+    }
+
     @Test
     public void testConsumeOneTaskEvent() throws Exception {
         Task event = Utils.unmarshalFrom(MINIMAL_TASK, Task.TYPE_REFERENCE);
@@ -343,7 +364,9 @@ public void onComplete() {
 
     @Test
     public void testConsumeAllStopsOnQueueClosed() throws Exception {
-        EventQueue queue = EventQueueUtil.getEventQueueBuilder().build().tap();
+        EventQueue queue = EventQueueUtil.getEventQueueBuilder()
+                .mainEventBus(mainEventBus)
+                .build().tap();
         EventConsumer consumer = new EventConsumer(queue);
 
         // Close the queue immediately
@@ -389,7 +412,9 @@ public void onComplete() {
 
     @Test
     public void testConsumeAllHandlesQueueClosedException() throws Exception {
-        EventQueue queue = EventQueueUtil.getEventQueueBuilder().build().tap();
+        EventQueue queue = EventQueueUtil.getEventQueueBuilder()
+                .mainEventBus(mainEventBus)
+                .build().tap();
         EventConsumer consumer = new EventConsumer(queue);
 
         // Add a message event (which will complete the stream)
@@ -447,7 +472,9 @@ public void onComplete() {
 
     @Test
     public void testConsumeAllTerminatesOnQueueClosedEvent() throws Exception {
-        EventQueue queue = EventQueueUtil.getEventQueueBuilder().build().tap();
+        EventQueue queue = EventQueueUtil.getEventQueueBuilder()
+                .mainEventBus(mainEventBus)
+                .build().tap();
         EventConsumer consumer = new EventConsumer(queue);
 
         // Enqueue a QueueClosedEvent (poison pill)

server-common/src/test/java/io/a2a/server/events/EventQueueTest.java

@@ -225,7 +225,9 @@ public void testCloseImmediatePropagationToChildren() throws Exception {
 
     @Test
     public void testEnqueueEventWhenClosed() throws Exception {
-        EventQueue mainQueue = EventQueueUtil.getEventQueueBuilder().build();
+        EventQueue mainQueue = EventQueueUtil.getEventQueueBuilder()
+                .mainEventBus(mainEventBus)
+                .build();
         EventQueue childQueue = mainQueue.tap();
         Event event = Utils.unmarshalFrom(MINIMAL_TASK, Task.TYPE_REFERENCE);
 
@@ -260,7 +262,9 @@ public void testDequeueEventWhenClosedAndEmpty() throws Exception {
 
     @Test
     public void testDequeueEventWhenClosedButHasEvents() throws Exception {
-        EventQueue mainQueue = EventQueueUtil.getEventQueueBuilder().build();
+        EventQueue mainQueue = EventQueueUtil.getEventQueueBuilder()
+                .mainEventBus(mainEventBus)
+                .build();
         EventQueue childQueue = mainQueue.tap();
         Event event = Utils.unmarshalFrom(MINIMAL_TASK, Task.TYPE_REFERENCE);
 

server-common/src/test/java/io/a2a/server/events/EventQueueUtil.java

@@ -1,28 +1,12 @@
 package io.a2a.server.events;
 
-import io.a2a.server.tasks.InMemoryTaskStore;
-import io.a2a.server.tasks.PushNotificationSender;
-import io.a2a.server.tasks.TaskStateProvider;
 import java.util.concurrent.atomic.AtomicInteger;
 
 public class EventQueueUtil {
     // Shared MainEventBus for all tests (to avoid creating one per test)
+    // Each test creates its own MainEventBusProcessor in @BeforeEach/@AfterEach for proper isolation
     private static final MainEventBus TEST_EVENT_BUS = new MainEventBus();
 
-    // Shared MainEventBusProcessor for all tests (automatically processes events)
-    private static final MainEventBusProcessor TEST_PROCESSOR;
-
-    static {
-        // Initialize and start the processor once for all tests
-        InMemoryTaskStore testTaskStore = new InMemoryTaskStore();
-        PushNotificationSender testPushSender = taskId -> {}; // No-op for tests
-        TEST_PROCESSOR = new MainEventBusProcessor(TEST_EVENT_BUS, testTaskStore, testPushSender);
-        TEST_PROCESSOR.start();  // Start background thread
-
-        // Register shutdown hook to stop processor
-        Runtime.getRuntime().addShutdownHook(new Thread(() -> TEST_PROCESSOR.stop()));
-    }
-
     // Counter for generating unique test taskIds
     private static final AtomicInteger TASK_ID_COUNTER = new AtomicInteger(0);
 
@@ -39,15 +23,6 @@ public static EventQueue.EventQueueBuilder getEventQueueBuilder() {
                 .taskId("test-task-" + TASK_ID_COUNTER.incrementAndGet());
     }
 
-    /**
-     * Get the shared test MainEventBus instance.
-     *
-     * @return the test MainEventBus
-     */
-    public static MainEventBus getTestEventBus() {
-        return TEST_EVENT_BUS;
-    }
-
     /**
      * Start a MainEventBusProcessor instance.
      *

server-common/src/test/java/io/a2a/server/tasks/TaskUpdaterTest.java

@@ -16,13 +16,16 @@
 import io.a2a.server.events.EventQueue;
 import io.a2a.server.events.EventQueueItem;
 import io.a2a.server.events.EventQueueUtil;
+import io.a2a.server.events.MainEventBus;
+import io.a2a.server.events.MainEventBusProcessor;
 import io.a2a.spec.Event;
 import io.a2a.spec.Message;
 import io.a2a.spec.Part;
 import io.a2a.spec.TaskArtifactUpdateEvent;
 import io.a2a.spec.TaskState;
 import io.a2a.spec.TaskStatusUpdateEvent;
 import io.a2a.spec.TextPart;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -39,21 +42,41 @@ public class TaskUpdaterTest {
 
     private static final List<Part<?>> SAMPLE_PARTS = List.of(new TextPart("Test message"));
 
+    private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = task -> {};
+
     EventQueue eventQueue;
+    private MainEventBus mainEventBus;
+    private MainEventBusProcessor mainEventBusProcessor;
     private TaskUpdater taskUpdater;
 
 
 
     @BeforeEach
     public void init() {
-        eventQueue = EventQueueUtil.getEventQueueBuilder().build().tap();
+        // Set up MainEventBus and processor for production-like test environment
+        InMemoryTaskStore taskStore = new InMemoryTaskStore();
+        mainEventBus = new MainEventBus();
+        mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER);
+        EventQueueUtil.start(mainEventBusProcessor);
+
+        eventQueue = EventQueueUtil.getEventQueueBuilder()
+                .taskId(TEST_TASK_ID)
+                .mainEventBus(mainEventBus)
+                .build().tap();
         RequestContext context = new RequestContext.Builder()
                 .setTaskId(TEST_TASK_ID)
                 .setContextId(TEST_TASK_CONTEXT_ID)
                 .build();
         taskUpdater = new TaskUpdater(context, eventQueue);
     }
 
+    @AfterEach
+    public void cleanup() {
+        if (mainEventBusProcessor != null) {
+            EventQueueUtil.stop(mainEventBusProcessor);
+        }
+    }
+
     @Test
     public void testAddArtifactWithCustomIdAndName() throws Exception {
         taskUpdater.addArtifact(SAMPLE_PARTS, "custom-artifact-id", "Custom Artifact", null);