From b4c95dd2192fa9871d269ccf438e7001d3c0fb38 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Wed, 28 Oct 2026 15:19:11 +0100 Subject: [PATCH] fix(server): align builder blocking timeouts with documented defaults Expose timeout overrides for manually wired request handlers and document their configuration. Extend blocking send regression coverage to include artifacts, omitted configuration, and fire-and-forget resubscription. This fixes #1247 --- docs/content/dev/server.md | 2 + .../DefaultRequestHandler.java | 61 +++++++++++++++-- .../DefaultRequestHandlerTest.java | 67 ++++++++++++++++++- 3 files changed, 123 insertions(+), 7 deletions(-) diff --git a/docs/content/dev/server.md b/docs/content/dev/server.md index 9702fd10e..b80227599 100644 --- a/docs/content/dev/server.md +++ b/docs/content/dev/server.md @@ -295,4 +295,6 @@ class A2AServerConfiguration { 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. +Manually built handlers use blocking timeouts of 30 seconds for agent execution, 5 seconds for event consumption, and 1 second for TaskStore reconciliation. Override these with `agentCompletionTimeoutSeconds(int)`, `consumptionCompletionTimeoutSeconds(int)`, and `reconciliationTimeoutSeconds(int)` on the builder. The `a2a.blocking.*` configuration properties apply to CDI-managed handlers; manual wiring uses the builder methods. + 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/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index ebda76a1c..aa99eab0a 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 @@ -153,9 +153,14 @@ *
  • Blocking (configuration.blocking=true): Client waits for first event or final task state
    • *
  • Streaming: Client receives events as they arrive via reactive streams
    • *
  • Both modes support fire-and-forget (agent continues after client disconnect)
    • - *
  • Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds}, + *
  • For CDI-managed handlers, configurable timeouts via {@code a2a.blocking.agent.timeout.seconds}, * {@code a2a.blocking.consumption.timeout.seconds}, and * {@code a2a.blocking.reconciliation.timeout.seconds}
    • + *
  • For manually built handlers, configure timeouts with + * {@link Builder#agentCompletionTimeoutSeconds(int)}, + * {@link Builder#consumptionCompletionTimeoutSeconds(int)}, and + * {@link Builder#reconciliationTimeoutSeconds(int)}. + * Both paths default to 30, 5, and 1 seconds, respectively.
    • * * *

    CDI Dependencies

    @@ -405,6 +410,9 @@ public static class Builder { private boolean authorizationRequired = true; private boolean populateReferredTasks; private boolean pushNotificationsEnabled = true; + private int agentCompletionTimeoutSeconds = 30; + private int consumptionCompletionTimeoutSeconds = 5; + private int reconciliationTimeoutSeconds = 1; public Builder agentExecutor(AgentExecutor agentExecutor) { this.agentExecutor = agentExecutor; @@ -441,6 +449,51 @@ public Builder eventConsumerExecutor(Executor eventConsumerExecutor) { return this; } + /** + * Sets how long a blocking call waits for agent execution. Defaults to 30 seconds. + * + * @param seconds timeout in seconds; zero skips waiting + * @return this builder for chaining + * @throws IllegalArgumentException if the timeout is negative + */ + public Builder agentCompletionTimeoutSeconds(int seconds) { + if (seconds < 0) { + throw new IllegalArgumentException("Agent completion timeout must not be negative"); + } + this.agentCompletionTimeoutSeconds = seconds; + return this; + } + + /** + * Sets how long a blocking call waits for event consumption. Defaults to 5 seconds. + * + * @param seconds timeout in seconds; zero skips waiting + * @return this builder for chaining + * @throws IllegalArgumentException if the timeout is negative + */ + public Builder consumptionCompletionTimeoutSeconds(int seconds) { + if (seconds < 0) { + throw new IllegalArgumentException("Consumption completion timeout must not be negative"); + } + this.consumptionCompletionTimeoutSeconds = seconds; + return this; + } + + /** + * Sets how long TaskStore reconciliation polls in blocking calls. Defaults to 1 second. + * + * @param seconds timeout in seconds; zero disables reconciliation polling + * @return this builder for chaining + * @throws IllegalArgumentException if the timeout is negative + */ + public Builder reconciliationTimeoutSeconds(int seconds) { + if (seconds < 0) { + throw new IllegalArgumentException("Reconciliation timeout must not be negative"); + } + this.reconciliationTimeoutSeconds = seconds; + return this; + } + public Builder authorizationProvider(@Nullable TaskAuthorizationProvider authorizationProvider) { this.authorizationProvider = authorizationProvider; return this; @@ -490,9 +543,9 @@ public DefaultRequestHandler build() { DefaultRequestHandler handler = new DefaultRequestHandler(agentExecutor, taskStore, queueManager, pushConfigStore, mainEventBusProcessor, executor, eventConsumerExecutor); - handler.agentCompletionTimeoutSeconds = 5; - handler.consumptionCompletionTimeoutSeconds = 2; - handler.reconciliationTimeoutSeconds = 1; + handler.agentCompletionTimeoutSeconds = agentCompletionTimeoutSeconds; + handler.consumptionCompletionTimeoutSeconds = consumptionCompletionTimeoutSeconds; + handler.reconciliationTimeoutSeconds = reconciliationTimeoutSeconds; handler.authorizationProvider = authorizationProvider; handler.agentExecutorRouter = agentExecutorRouter; handler.authorizationRequired = authorizationRequired; 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 77f4bb9b4..eac9f77a4 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 @@ -203,6 +203,44 @@ void testInitConfigReadsBlockingTimeouts() { assertEquals(7, handler.reconciliationTimeoutSeconds); } + @Test + void testBuilderUsesDocumentedBlockingTimeoutDefaults() { + DefaultRequestHandler handler = (DefaultRequestHandler) requestHandler; + + assertEquals(30, handler.agentCompletionTimeoutSeconds); + assertEquals(5, handler.consumptionCompletionTimeoutSeconds); + assertEquals(1, handler.reconciliationTimeoutSeconds); + } + + @ParameterizedTest + @CsvSource({"41,8,3", "0,0,0"}) + void testBuilderOverridesBlockingTimeouts(int agentTimeout, int consumptionTimeout, int reconciliationTimeout) { + DefaultRequestHandler handler = DefaultRequestHandler.builder() + .agentExecutor(executor) + .taskStore(taskStore) + .queueManager(queueManager) + .mainEventBusProcessor(mainEventBusProcessor) + .executor(internalExecutor) + .eventConsumerExecutor(internalExecutor) + .agentCompletionTimeoutSeconds(agentTimeout) + .consumptionCompletionTimeoutSeconds(consumptionTimeout) + .reconciliationTimeoutSeconds(reconciliationTimeout) + .build(); + + assertEquals(agentTimeout, handler.agentCompletionTimeoutSeconds); + assertEquals(consumptionTimeout, handler.consumptionCompletionTimeoutSeconds); + assertEquals(reconciliationTimeout, handler.reconciliationTimeoutSeconds); + } + + @Test + void testBuilderRejectsNegativeBlockingTimeouts() { + DefaultRequestHandler.Builder builder = DefaultRequestHandler.builder(); + + assertThrows(IllegalArgumentException.class, () -> builder.agentCompletionTimeoutSeconds(-1)); + assertThrows(IllegalArgumentException.class, () -> builder.consumptionCompletionTimeoutSeconds(-1)); + assertThrows(IllegalArgumentException.class, () -> builder.reconciliationTimeoutSeconds(-1)); + } + @Test void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exception { InMemoryTaskStore manualTaskStore = new InMemoryTaskStore(); @@ -1758,8 +1796,9 @@ public void cancel(RequestContext context, AgentEmitter emitter) { * must still reach the queue, which was told to expect it; otherwise the consumer waits out its * grace period for it and the call fails or returns late. */ - @Test - void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate() throws Exception { + @ParameterizedTest + @CsvSource({"true", "false"}) + void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate(boolean omitConfiguration) throws Exception { AtomicReference callerQueue = new AtomicReference<>(); useQueueManager(new InMemoryQueueManager((TaskStateProvider) taskStore, mainEventBus) { @Override @@ -1772,15 +1811,37 @@ public EventQueue createOrTap(String taskId) { AtomicBoolean callerQueueClosedInTime = new AtomicBoolean(); holdEventsAfterWorkingUntilClosed(callerQueue, callerQueueClosedInTime); agentExecutorExecute = (context, emitter) -> { + emitter.submit(); emitter.startWork(); + emitter.addArtifact(List.of(new TextPart("answer")), "answer-id", "answer", null); emitter.complete(); }; - EventKind result = requestHandler.onMessageSend(blockingParams("msg-blocking-final-event"), NULL_CONTEXT); + MessageSendParams params = blockingParams("msg-blocking-final-event"); + if (omitConfiguration) { + params = MessageSendParams.builder().message(params.message()).build(); + } + EventKind result = requestHandler.onMessageSend(params, NULL_CONTEXT); assertTrue(callerQueueClosedInTime.get(), "the caller's queue was not closed while the final event was held"); Task task = assertInstanceOf(Task.class, result); assertEquals(TaskState.TASK_STATE_COMPLETED, task.status().state()); + assertEquals(1, task.artifacts().size()); + assertEquals("answer-id", task.artifacts().get(0).artifactId()); + assertEquals("answer", assertInstanceOf(TextPart.class, task.artifacts().get(0).parts().get(0)).text()); + } + + @Test + void blockingSendWithoutFinalEventReturnsWorkingTaskAndAllowsResubscription() throws Exception { + agentExecutorExecute = (context, emitter) -> emitter.startWork(); + + EventKind result = requestHandler.onMessageSend(blockingParams("msg-fire-and-forget"), NULL_CONTEXT); + + Task task = assertInstanceOf(Task.class, result); + assertEquals(TaskState.TASK_STATE_WORKING, task.status().state()); + EventQueue resubscribedQueue = queueManager.tap(task.id()); + assertNotNull(resubscribedQueue); + resubscribedQueue.close(); } private void useQueueManager(InMemoryQueueManager manager) {