From 12bb82e62d9b2417427c9eca9031b441da01d3c1 Mon Sep 17 00:00:00 2001 From: 014-code <2402143478@qq.com> Date: Sat, 25 Jul 2026 22:37:02 +0800 Subject: [PATCH 1/4] fix(server): start event bus processor for manual wiring Ensure MainEventBusProcessor.ensureStarted() starts manually constructed processors, and call it from DefaultRequestHandler.create(...). This prevents manually wired integrations from leaving events stuck on the main event bus. Add a regression test for the manual wiring path that returns a Message response. This fixes #995 --- .../server/events/MainEventBusProcessor.java | 18 +++++++--- .../DefaultRequestHandlerTest.java | 35 +++++++++++++++++++ 2 files changed, 48 insertions(+), 5 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java index cd293f672..f27e1904e 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java @@ -158,7 +158,12 @@ int pushNotificationChainCount() { @SuppressWarnings("NullAway.Init") @PostConstruct - void start() { + synchronized void start() { + if (processorThread != null) { + LOGGER.debug("MainEventBusProcessor already started"); + return; + } + running = true; processorThread = new Thread(this, "MainEventBusProcessor"); processorThread.setDaemon(true); // Allow JVM to exit even if this thread is running processorThread.start(); @@ -166,15 +171,18 @@ void start() { } /** - * No-op method to force CDI proxy resolution and ensure @PostConstruct has been called. - * Called by MainEventBusProcessorInitializer during application startup. + * Ensures the background processor thread has been started. + *

+ * In CDI runtimes, this forces proxy resolution and {@link PostConstruct}. For manual + * wiring, this starts the processor directly. + *

*/ public void ensureStarted() { - // Method intentionally empty - just forces proxy resolution + start(); } @PreDestroy - void stop() { + synchronized void stop() { LOGGER.info("MainEventBusProcessor stopping..."); running = false; if (processorThread != null) { diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index 53b25b1a1..7238a12bf 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -186,6 +186,41 @@ void testInitConfigReadsBlockingTimeouts() { assertEquals(7, handler.reconciliationTimeoutSeconds); } + @Test + void testCreateStartsManuallyConstructedMainEventBusProcessor() throws Exception { + InMemoryTaskStore manualTaskStore = new InMemoryTaskStore(); + PushNotificationConfigStore manualPushConfigStore = new InMemoryPushNotificationConfigStore(); + MainEventBus manualMainEventBus = new MainEventBus(); + InMemoryQueueManager manualQueueManager = new InMemoryQueueManager(manualTaskStore, manualMainEventBus); + MainEventBusProcessor manualProcessor = new MainEventBusProcessor( + manualMainEventBus, manualTaskStore, NOOP_PUSHNOTIFICATION_SENDER, manualQueueManager); + + try { + AgentExecutor manualAgentExecutor = new AgentExecutor() { + @Override + public void execute(RequestContext context, AgentEmitter agentEmitter) { + agentEmitter.sendMessage("manual lifecycle response"); + } + + @Override + public void cancel(RequestContext context, AgentEmitter agentEmitter) { + throw new AssertionError("Cancel should not be invoked"); + } + }; + RequestHandler manualRequestHandler = DefaultRequestHandler.create( + manualAgentExecutor, manualTaskStore, manualQueueManager, manualPushConfigStore, + manualProcessor, internalExecutor, internalExecutor); + + EventKind eventKind = manualRequestHandler.onMessageSend( + MessageSendParams.builder().message(MESSAGE).build(), NULL_CONTEXT); + + assertInstanceOf(Message.class, eventKind); + assertEquals(Message.Role.ROLE_AGENT, ((Message) eventKind).role()); + } finally { + EventQueueUtil.stop(manualProcessor); + } + } + /** * Test 1: Non-streaming AUTH_REQUIRED returns immediately while agent continues. * Verifies: From e2d60112e7bcc1d9c3244a588cb064eaa224df16 Mon Sep 17 00:00:00 2001 From: 014-code <2402143478@qq.com> Date: Mon, 27 Jul 2026 20:43:17 +0800 Subject: [PATCH 2/4] fix(server): cover direct request handler wiring Start the main event bus processor from the public request handler constructor and verify the direct manual wiring path.\n\nThis fixes #995 --- .../sdk/server/events/MainEventBusProcessor.java | 2 +- .../sdk/server/requesthandlers/DefaultRequestHandler.java | 1 + .../server/requesthandlers/DefaultRequestHandlerTest.java | 7 +++++-- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java index f27e1904e..bfbe54bc9 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java @@ -159,7 +159,7 @@ int pushNotificationChainCount() { @SuppressWarnings("NullAway.Init") @PostConstruct synchronized void start() { - if (processorThread != null) { + if (processorThread != null && processorThread.isAlive()) { LOGGER.debug("MainEventBusProcessor already started"); return; } diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 7c97cc003..606f9481a 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -326,6 +326,7 @@ public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore, // I am unsure about the correct scope. // Also reworked to make a Supplier since otherwise the builder gets polluted with wrong tasks this.requestContextBuilder = () -> new SimpleRequestContextBuilder(taskStore, false, null); + this.mainEventBusProcessor.start(); } @SuppressWarnings("NullAway.Init") diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index 7238a12bf..c8894456d 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -187,7 +187,7 @@ void testInitConfigReadsBlockingTimeouts() { } @Test - void testCreateStartsManuallyConstructedMainEventBusProcessor() throws Exception { + void testConstructorStartsManuallyConstructedMainEventBusProcessor() throws Exception { InMemoryTaskStore manualTaskStore = new InMemoryTaskStore(); PushNotificationConfigStore manualPushConfigStore = new InMemoryPushNotificationConfigStore(); MainEventBus manualMainEventBus = new MainEventBus(); @@ -207,9 +207,12 @@ public void cancel(RequestContext context, AgentEmitter agentEmitter) { throw new AssertionError("Cancel should not be invoked"); } }; - RequestHandler manualRequestHandler = DefaultRequestHandler.create( + DefaultRequestHandler manualRequestHandler = new DefaultRequestHandler( manualAgentExecutor, manualTaskStore, manualQueueManager, manualPushConfigStore, manualProcessor, internalExecutor, internalExecutor); + manualRequestHandler.agentCompletionTimeoutSeconds = 5; + manualRequestHandler.consumptionCompletionTimeoutSeconds = 2; + manualRequestHandler.reconciliationTimeoutSeconds = 1; EventKind eventKind = manualRequestHandler.onMessageSend( MessageSendParams.builder().message(MESSAGE).build(), NULL_CONTEXT); From 4b6eac768e8f724ab901df6b839ed166765b9876 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Fri, 25 Sep 2026 13:50:03 +0100 Subject: [PATCH 3/4] fix(server): expose event processor startup --- docs/content/dev/server.md | 32 +++++++++++++++++++ .../server/events/MainEventBusProcessor.java | 13 +------- .../MainEventBusProcessorInitializer.java | 6 ++-- 3 files changed, 35 insertions(+), 16 deletions(-) diff --git a/docs/content/dev/server.md b/docs/content/dev/server.md index 72edda957..b6d6a225a 100644 --- a/docs/content/dev/server.md +++ b/docs/content/dev/server.md @@ -247,4 +247,36 @@ See [Backward Compatibility](compatibility) for multi-version modules, version r - **Quarkus** — Reference implementations are Quarkus-based (JSON-RPC, gRPC, REST) - **Jakarta EE** — [a2a-jakarta](https://github.com/wildfly-extras/a2a-jakarta) works with any Jakarta EE Web Profile runtime +### Manual Wiring + +For example, when wiring the server components as Spring beans, expose the event processor as a bean and pass it to the request handler. The handler starts the processor when it is constructed, including when the processor was created manually: + +```java +@Configuration +class A2AServerConfiguration { + + @Bean + MainEventBus mainEventBus() { + return new MainEventBus(); + } + + @Bean + MainEventBusProcessor mainEventBusProcessor(MainEventBus mainEventBus, TaskStore taskStore, + PushNotificationSender pushSender, QueueManager queueManager) { + return new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager); + } + + @Bean + RequestHandler requestHandler(AgentExecutor agentExecutor, TaskStore taskStore, + QueueManager queueManager, PushNotificationConfigStore pushConfigStore, + MainEventBusProcessor mainEventBusProcessor, + @Qualifier("a2aInternal") Executor executor) { + return new DefaultRequestHandler(agentExecutor, taskStore, queueManager, pushConfigStore, + mainEventBusProcessor, executor, executor); + } +} +``` + +This assumes the other method parameters are also registered as Spring beans. The same executor is passed for agent execution and event consumption, as in this example; applications can provide separate executors when needed. + See [CONTRIBUTING_INTEGRATIONS.md](https://github.com/a2aproject/a2a-java/blob/main/CONTRIBUTING_INTEGRATIONS.md) to submit your own integration. diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java index bfbe54bc9..9cb2632aa 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java @@ -158,7 +158,7 @@ int pushNotificationChainCount() { @SuppressWarnings("NullAway.Init") @PostConstruct - synchronized void start() { + public synchronized void start() { if (processorThread != null && processorThread.isAlive()) { LOGGER.debug("MainEventBusProcessor already started"); return; @@ -170,17 +170,6 @@ synchronized void start() { LOGGER.info("MainEventBusProcessor started"); } - /** - * Ensures the background processor thread has been started. - *

- * In CDI runtimes, this forces proxy resolution and {@link PostConstruct}. For manual - * wiring, this starts the processor directly. - *

- */ - public void ensureStarted() { - start(); - } - @PreDestroy synchronized void stop() { LOGGER.info("MainEventBusProcessor stopping..."); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java index 707b04d1f..615cd4c46 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java @@ -27,14 +27,12 @@ public class MainEventBusProcessorInitializer { /** * Observes ApplicationScoped initialization to force eager creation of MainEventBusProcessor. - * The injection of MainEventBusProcessor in this bean triggers its creation, and calling - * ensureStarted() forces the CDI proxy to be resolved, which ensures @PostConstruct has been + * Calling start() forces the CDI proxy to be resolved, which ensures @PostConstruct has been * called and the background thread is running. */ void onStart(@Observes @Initialized(ApplicationScoped.class) Object event) { if (processor != null) { - // Force proxy resolution to ensure @PostConstruct has been called - processor.ensureStarted(); + processor.start(); LOGGER.info("MainEventBusProcessor initialized and started"); } else { LOGGER.error("MainEventBusProcessor is null - initialization failed!"); From 088bc772c0462b9181fd5bfc40ab44981c16939a Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Fri, 25 Sep 2026 13:59:17 +0100 Subject: [PATCH 4/4] fix(server): cover builder-based manual processor startup --- docs/content/dev/server.md | 23 +++++++++++++++---- .../server/events/MainEventBusProcessor.java | 3 +++ .../DefaultRequestHandler.java | 4 ++++ .../DefaultRequestHandlerTest.java | 15 ++++++++---- 4 files changed, 37 insertions(+), 8 deletions(-) diff --git a/docs/content/dev/server.md b/docs/content/dev/server.md index b6d6a225a..91d1a3e09 100644 --- a/docs/content/dev/server.md +++ b/docs/content/dev/server.md @@ -266,17 +266,32 @@ class A2AServerConfiguration { return new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager); } + @Bean(destroyMethod = "shutdown") + ExecutorService a2aEventConsumerExecutor() { + return Executors.newCachedThreadPool(); + } + @Bean RequestHandler requestHandler(AgentExecutor agentExecutor, TaskStore taskStore, QueueManager queueManager, PushNotificationConfigStore pushConfigStore, MainEventBusProcessor mainEventBusProcessor, - @Qualifier("a2aInternal") Executor executor) { - return new DefaultRequestHandler(agentExecutor, taskStore, queueManager, pushConfigStore, - mainEventBusProcessor, executor, executor); + TaskAuthorizationProvider authorizationProvider, + @Qualifier("a2aInternalExecutor") Executor executor, + @Qualifier("a2aEventConsumerExecutor") Executor eventConsumerExecutor) { + return DefaultRequestHandler.builder() + .agentExecutor(agentExecutor) + .taskStore(taskStore) + .queueManager(queueManager) + .pushConfigStore(pushConfigStore) + .mainEventBusProcessor(mainEventBusProcessor) + .authorizationProvider(authorizationProvider) + .executor(executor) + .eventConsumerExecutor(eventConsumerExecutor) + .build(); } } ``` -This assumes the other method parameters are also registered as Spring beans. The same executor is passed for agent execution and event consumption, as in this example; applications can provide separate executors when needed. +Register the remaining dependencies, including an `a2aInternalExecutor` bean and a `TaskAuthorizationProvider`, as Spring beans. Keep the cached event consumer executor separate from the agent executor: blocking event polling can exhaust a bounded pool under load. See [CONTRIBUTING_INTEGRATIONS.md](https://github.com/a2aproject/a2a-java/blob/main/CONTRIBUTING_INTEGRATIONS.md) to submit your own integration. diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java index 9cb2632aa..c0bf0e92b 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java @@ -156,6 +156,9 @@ int pushNotificationChainCount() { return pushNotificationChains.size(); } + /** + * Starts the background processor thread if it is not already running. + */ @SuppressWarnings("NullAway.Init") @PostConstruct public synchronized void start() { diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 606f9481a..4388b0443 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -308,6 +308,10 @@ protected DefaultRequestHandler() { this.eventConsumerExecutor = null; } + /** + * Constructor used by CDI to create the request handler. Its signature may change between + * releases; application code should use {@link #builder()} to configure and create a handler. + */ @Inject public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore, QueueManager queueManager, PushNotificationConfigStore pushConfigStore, diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index c8894456d..1e729bd20 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -187,7 +187,7 @@ void testInitConfigReadsBlockingTimeouts() { } @Test - void testConstructorStartsManuallyConstructedMainEventBusProcessor() throws Exception { + void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exception { InMemoryTaskStore manualTaskStore = new InMemoryTaskStore(); PushNotificationConfigStore manualPushConfigStore = new InMemoryPushNotificationConfigStore(); MainEventBus manualMainEventBus = new MainEventBus(); @@ -207,9 +207,16 @@ public void cancel(RequestContext context, AgentEmitter agentEmitter) { throw new AssertionError("Cancel should not be invoked"); } }; - DefaultRequestHandler manualRequestHandler = new DefaultRequestHandler( - manualAgentExecutor, manualTaskStore, manualQueueManager, manualPushConfigStore, - manualProcessor, internalExecutor, internalExecutor); + DefaultRequestHandler manualRequestHandler = DefaultRequestHandler.builder() + .agentExecutor(manualAgentExecutor) + .taskStore(manualTaskStore) + .queueManager(manualQueueManager) + .pushConfigStore(manualPushConfigStore) + .mainEventBusProcessor(manualProcessor) + .executor(internalExecutor) + .eventConsumerExecutor(internalExecutor) + .authorizationRequired(false) + .build(); manualRequestHandler.agentCompletionTimeoutSeconds = 5; manualRequestHandler.consumptionCompletionTimeoutSeconds = 2; manualRequestHandler.reconciliationTimeoutSeconds = 1;