Repository navigation
fix: harden event consumer and queue - #1040
Conversation
kabir
left a comment
kabir
left a comment
There was a problem hiding this comment.
Thanks for the PR — changes #3 (bounded ChildQueue) and #4 (semaphore leak on submit failure) are solid bug fixes. Change #2 (configurable flush delay) is reasonable though I'm not sure we need the knob yet.
However, I have a concern with change #1 (removing Message as a stream-terminating event).
The A2A spec (§3.1.2, Send Streaming Message) is explicit about this:
"If the agent returns a
Message, the stream MUST contain exactly oneMessageobject and then close immediately. No task tracking or updates are provided."
Message and Task are mutually exclusive response types for a single message/stream request. There is no concept of an "intermediate Message" in a stream — a Message is the complete response. The current code treating Message as isFinalEvent = true is correct per spec.
Removing this would violate the spec by keeping a Message-only stream open unnecessarily, only for it to eventually drain via the 1.5s agent-completed grace period instead of closing immediately as required.
The observation that MainQueue.isFinalEvent doesn't consider Message is not an inconsistency — that method tracks whether a task reached a terminal state (for persistence/distribution purposes), which is a different concern from whether the client-facing stream should close.
Could you share the scenario that motivated this change? If there's a concrete case where a Message arrives before a Task in the same stream, that would be worth discussing — but per the spec as written, that shouldn't happen.
Revert the change that treated Message events as non-terminating. Per the A2A spec (Send Streaming Message), a Message is the complete response and the stream must close after delivering it. The EventConsumer serves both message/stream and task subscriptions, and the spec-conformant behavior for message/stream takes precedence. The BUG-26-motivated test asserting the opposite is removed. Keep the other three changes: configurable buffer-flush delay, bounded ChildQueue, and the semaphore permit leak fix.
|
Thanks for the detailed review — you're right about the spec. You correctly identified that §3.1.2 (Send Streaming Message) requires a The remaining three changes are kept as-is:
Thanks again for catching this. |
The revert of the "Message is not stream-terminating" change left behind a contradictory NOTE comment and didn't restore testConsumeMessageEvents. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
fix: remove stale comment and restore deleted test from revert
- Fix import ordering of TaskQueryParams in DefaultRequestHandlerTest - Remove fragile reflection-based test for negative historyLength guard (the two public-API tests already exercise the limitTaskHistory path) - Trim verbose 4-line comment on negative historyLength guard to one line - Fix Javadoc on bufferFlushDelayMs(): negative values are clamped to 0, not returned as the default - Replace fully-qualified class names with imports in EventQueueTest Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
What changed
1. Buffer-flush delay is configurable instead of hardcoded
Problem:
EventConsumerblocked its polling thread with a hardcodedThread.sleep(150)(BUFFER_FLUSH_DELAY_MS) before completing a stream. The value was not tunable for transports where the flush needs differ.Fix (EventConsumer.java):
a2a.eventconsumer.bufferFlushDelayMssystem property (default 150 ms, clamped to >= 0; invalid values fall back to the default). Semantics are preserved: the delay still runs only once per stream, after the final event is sent, to let the SSE write callback fire beforetube.complete().bufferFlushDelayMs()accessor and tests (testBufferFlushDelayMsDefaultsTo150,testBufferFlushDelayMsReadsConfiguredValue,testBufferFlushDelayMsRejectsInvalidValues).Behavior change: none unless the system property is set.
2. ChildQueue is bounded by the parent queue size
Problem:
ChildQueuebuilt itsLinkedBlockingDequewithout a capacity, soqueue.offer()always returnedtrueand a slow subscriber could grow the deque without limit (the "queue is full" → immediate-close path was dead code).Fix (server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java):
ChildQueuenow constructs its deque with the parent'squeueSize(new LinkedBlockingDeque<>(parent.getQueueSize())), restoring the intended overflow → immediate-close behavior for slow consumers.Fix (server-common/src/test/java/org/a2aproject/sdk/server/events/EventQueueTest.java):
testChildQueueIsBoundedByParentQueueSizeverifies the child deque'sremainingCapacityequals the configured queue size.3. Semaphore permit is released when MainEventBus.submit fails
Problem:
MainQueue.enqueueItemacquires a semaphore permit, then callsmainEventBus.submit(...). Ifsubmitthrows (e.g. interrupted), the permit was never released — the normal release only happens inMainEventBusProcessor.processEvent'sfinally, which never runs for an unsubmitted event. Repeated failures would leak all permits and permanently block event processing for the task.Fix (EventQueue.java):
mainEventBus.submit(...)in try/catch: on aRuntimeExceptionthe acquired permit is released before rethrowing. The success path is unchanged (release still happens once inMainEventBusProcessor).Fix (EventQueueTest.java):
testSemaphorePermitReleasedWhenSubmitFailsmocksMainEventBus.submitto throw and assertsmainQueue.size()returns to 0 (no leaked permit).Testing
mvn -pl server-common test -Dtest=EventConsumerTest,EventQueueTest— 50 tests run, 0 failures, 0 errors, 0 skipped (BUILD SUCCESS).Message-terminates-stream behavior is preserved (per A2A spec §3.1.2), and the remaining changes (configurable flush delay, bounded ChildQueue, semaphore release) do not alter stream semantics.