Skip to content

Commit 4aeac45

Browse files
authored
fix(server): align builder blocking timeouts with documented defaults (#1249)
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
1 parent e90f8d9 commit 4aeac45

3 files changed

Lines changed: 123 additions & 7 deletions

File tree

‎docs/content/dev/server.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -295,4 +295,6 @@ class A2AServerConfiguration {
295295

296296
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.
297297

298+
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.
299+
298300
See [CONTRIBUTING_INTEGRATIONS.md](https://github.com/a2aproject/a2a-java/blob/main/CONTRIBUTING_INTEGRATIONS.md) to submit your own integration.

‎server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java‎

Lines changed: 57 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -153,9 +153,14 @@
153153
* <li><b>Blocking (configuration.blocking=true):</b> Client waits for first event or final task state</li>
154154
* <li><b>Streaming:</b> Client receives events as they arrive via reactive streams</li>
155155
* <li>Both modes support fire-and-forget (agent continues after client disconnect)</li>
156-
* <li>Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds},
156+
* <li>For CDI-managed handlers, configurable timeouts via {@code a2a.blocking.agent.timeout.seconds},
157157
* {@code a2a.blocking.consumption.timeout.seconds}, and
158158
* {@code a2a.blocking.reconciliation.timeout.seconds}</li>
159+
* <li>For manually built handlers, configure timeouts with
160+
* {@link Builder#agentCompletionTimeoutSeconds(int)},
161+
* {@link Builder#consumptionCompletionTimeoutSeconds(int)}, and
162+
* {@link Builder#reconciliationTimeoutSeconds(int)}.
163+
* Both paths default to 30, 5, and 1 seconds, respectively.</li>
159164
* </ul>
160165
*
161166
* <h2>CDI Dependencies</h2>
@@ -405,6 +410,9 @@ public static class Builder {
405410
private boolean authorizationRequired = true;
406411
private boolean populateReferredTasks;
407412
private boolean pushNotificationsEnabled = true;
413+
private int agentCompletionTimeoutSeconds = 30;
414+
private int consumptionCompletionTimeoutSeconds = 5;
415+
private int reconciliationTimeoutSeconds = 1;
408416

409417
public Builder agentExecutor(AgentExecutor agentExecutor) {
410418
this.agentExecutor = agentExecutor;
@@ -441,6 +449,51 @@ public Builder eventConsumerExecutor(Executor eventConsumerExecutor) {
441449
return this;
442450
}
443451

452+
/**
453+
* Sets how long a blocking call waits for agent execution. Defaults to 30 seconds.
454+
*
455+
* @param seconds timeout in seconds; zero skips waiting
456+
* @return this builder for chaining
457+
* @throws IllegalArgumentException if the timeout is negative
458+
*/
459+
public Builder agentCompletionTimeoutSeconds(int seconds) {
460+
if (seconds < 0) {
461+
throw new IllegalArgumentException("Agent completion timeout must not be negative");
462+
}
463+
this.agentCompletionTimeoutSeconds = seconds;
464+
return this;
465+
}
466+
467+
/**
468+
* Sets how long a blocking call waits for event consumption. Defaults to 5 seconds.
469+
*
470+
* @param seconds timeout in seconds; zero skips waiting
471+
* @return this builder for chaining
472+
* @throws IllegalArgumentException if the timeout is negative
473+
*/
474+
public Builder consumptionCompletionTimeoutSeconds(int seconds) {
475+
if (seconds < 0) {
476+
throw new IllegalArgumentException("Consumption completion timeout must not be negative");
477+
}
478+
this.consumptionCompletionTimeoutSeconds = seconds;
479+
return this;
480+
}
481+
482+
/**
483+
* Sets how long TaskStore reconciliation polls in blocking calls. Defaults to 1 second.
484+
*
485+
* @param seconds timeout in seconds; zero disables reconciliation polling
486+
* @return this builder for chaining
487+
* @throws IllegalArgumentException if the timeout is negative
488+
*/
489+
public Builder reconciliationTimeoutSeconds(int seconds) {
490+
if (seconds < 0) {
491+
throw new IllegalArgumentException("Reconciliation timeout must not be negative");
492+
}
493+
this.reconciliationTimeoutSeconds = seconds;
494+
return this;
495+
}
496+
444497
public Builder authorizationProvider(@Nullable TaskAuthorizationProvider authorizationProvider) {
445498
this.authorizationProvider = authorizationProvider;
446499
return this;
@@ -490,9 +543,9 @@ public DefaultRequestHandler build() {
490543
DefaultRequestHandler handler =
491544
new DefaultRequestHandler(agentExecutor, taskStore, queueManager, pushConfigStore,
492545
mainEventBusProcessor, executor, eventConsumerExecutor);
493-
handler.agentCompletionTimeoutSeconds = 5;
494-
handler.consumptionCompletionTimeoutSeconds = 2;
495-
handler.reconciliationTimeoutSeconds = 1;
546+
handler.agentCompletionTimeoutSeconds = agentCompletionTimeoutSeconds;
547+
handler.consumptionCompletionTimeoutSeconds = consumptionCompletionTimeoutSeconds;
548+
handler.reconciliationTimeoutSeconds = reconciliationTimeoutSeconds;
496549
handler.authorizationProvider = authorizationProvider;
497550
handler.agentExecutorRouter = agentExecutorRouter;
498551
handler.authorizationRequired = authorizationRequired;

‎server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java‎

Lines changed: 64 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -203,6 +203,44 @@ void testInitConfigReadsBlockingTimeouts() {
203203
assertEquals(7, handler.reconciliationTimeoutSeconds);
204204
}
205205

206+
@Test
207+
void testBuilderUsesDocumentedBlockingTimeoutDefaults() {
208+
DefaultRequestHandler handler = (DefaultRequestHandler) requestHandler;
209+
210+
assertEquals(30, handler.agentCompletionTimeoutSeconds);
211+
assertEquals(5, handler.consumptionCompletionTimeoutSeconds);
212+
assertEquals(1, handler.reconciliationTimeoutSeconds);
213+
}
214+
215+
@ParameterizedTest
216+
@CsvSource({"41,8,3", "0,0,0"})
217+
void testBuilderOverridesBlockingTimeouts(int agentTimeout, int consumptionTimeout, int reconciliationTimeout) {
218+
DefaultRequestHandler handler = DefaultRequestHandler.builder()
219+
.agentExecutor(executor)
220+
.taskStore(taskStore)
221+
.queueManager(queueManager)
222+
.mainEventBusProcessor(mainEventBusProcessor)
223+
.executor(internalExecutor)
224+
.eventConsumerExecutor(internalExecutor)
225+
.agentCompletionTimeoutSeconds(agentTimeout)
226+
.consumptionCompletionTimeoutSeconds(consumptionTimeout)
227+
.reconciliationTimeoutSeconds(reconciliationTimeout)
228+
.build();
229+
230+
assertEquals(agentTimeout, handler.agentCompletionTimeoutSeconds);
231+
assertEquals(consumptionTimeout, handler.consumptionCompletionTimeoutSeconds);
232+
assertEquals(reconciliationTimeout, handler.reconciliationTimeoutSeconds);
233+
}
234+
235+
@Test
236+
void testBuilderRejectsNegativeBlockingTimeouts() {
237+
DefaultRequestHandler.Builder builder = DefaultRequestHandler.builder();
238+
239+
assertThrows(IllegalArgumentException.class, () -> builder.agentCompletionTimeoutSeconds(-1));
240+
assertThrows(IllegalArgumentException.class, () -> builder.consumptionCompletionTimeoutSeconds(-1));
241+
assertThrows(IllegalArgumentException.class, () -> builder.reconciliationTimeoutSeconds(-1));
242+
}
243+
206244
@Test
207245
void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exception {
208246
InMemoryTaskStore manualTaskStore = new InMemoryTaskStore();
@@ -1758,8 +1796,9 @@ public void cancel(RequestContext context, AgentEmitter emitter) {
17581796
* must still reach the queue, which was told to expect it; otherwise the consumer waits out its
17591797
* grace period for it and the call fails or returns late.
17601798
*/
1761-
@Test
1762-
void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate() throws Exception {
1799+
@ParameterizedTest
1800+
@CsvSource({"true", "false"})
1801+
void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate(boolean omitConfiguration) throws Exception {
17631802
AtomicReference<EventQueue> callerQueue = new AtomicReference<>();
17641803
useQueueManager(new InMemoryQueueManager((TaskStateProvider) taskStore, mainEventBus) {
17651804
@Override
@@ -1772,15 +1811,37 @@ public EventQueue createOrTap(String taskId) {
17721811
AtomicBoolean callerQueueClosedInTime = new AtomicBoolean();
17731812
holdEventsAfterWorkingUntilClosed(callerQueue, callerQueueClosedInTime);
17741813
agentExecutorExecute = (context, emitter) -> {
1814+
emitter.submit();
17751815
emitter.startWork();
1816+
emitter.addArtifact(List.of(new TextPart("answer")), "answer-id", "answer", null);
17761817
emitter.complete();
17771818
};
17781819

1779-
EventKind result = requestHandler.onMessageSend(blockingParams("msg-blocking-final-event"), NULL_CONTEXT);
1820+
MessageSendParams params = blockingParams("msg-blocking-final-event");
1821+
if (omitConfiguration) {
1822+
params = MessageSendParams.builder().message(params.message()).build();
1823+
}
1824+
EventKind result = requestHandler.onMessageSend(params, NULL_CONTEXT);
17801825

17811826
assertTrue(callerQueueClosedInTime.get(), "the caller's queue was not closed while the final event was held");
17821827
Task task = assertInstanceOf(Task.class, result);
17831828
assertEquals(TaskState.TASK_STATE_COMPLETED, task.status().state());
1829+
assertEquals(1, task.artifacts().size());
1830+
assertEquals("answer-id", task.artifacts().get(0).artifactId());
1831+
assertEquals("answer", assertInstanceOf(TextPart.class, task.artifacts().get(0).parts().get(0)).text());
1832+
}
1833+
1834+
@Test
1835+
void blockingSendWithoutFinalEventReturnsWorkingTaskAndAllowsResubscription() throws Exception {
1836+
agentExecutorExecute = (context, emitter) -> emitter.startWork();
1837+
1838+
EventKind result = requestHandler.onMessageSend(blockingParams("msg-fire-and-forget"), NULL_CONTEXT);
1839+
1840+
Task task = assertInstanceOf(Task.class, result);
1841+
assertEquals(TaskState.TASK_STATE_WORKING, task.status().state());
1842+
EventQueue resubscribedQueue = queueManager.tap(task.id());
1843+
assertNotNull(resubscribedQueue);
1844+
resubscribedQueue.close();
17841845
}
17851846

17861847
private void useQueueManager(InMemoryQueueManager manager) {

0 commit comments

Comments
 (0)