What happened?
Calling EventQueueSink.close() more than once is functionally safe, but the
second close exports an OpenTelemetry ERROR span for
EventQueueSource.remove_sink.
EventQueueSink.close() documents that repeated calls are safe and suppresses
KeyError around remove_sink(). However, EventQueueSource is decorated
with @trace_class, so remove_sink() is traced separately. Its
set.remove() raises before the caller suppresses the exception, and the
telemetry wrapper records the exception and sets the span status to ERROR.
This produces a false error observation even though cleanup completes normally.
If the second cleanup happens after the request, the late span can also extend
the apparent trace latency.
Minimal reproduction
source = EventQueueSource(create_default_sink=False)
sink = await source.tap()
await sink.close(immediate=True)
await sink.close(immediate=True)
await source.close(immediate=True)
With A2A OpenTelemetry instrumentation enabled, the second call records:
a2a.server.events.event_queue_v2.EventQueueSource.remove_sink
KeyError: <EventQueueSink ...>
status: ERROR
The application does not receive the exception because close() suppresses it.
Expected behavior
Repeated or concurrent sink cleanup should not raise an expected exception
inside the traced remove_sink() method, and no ERROR span should be
exported for a successful idempotent close.
Proposed fix
Use set.discard() when detaching a sink. It preserves removal when the sink is
present and does nothing when cleanup already removed it.
Environment
a2a-sdk==1.1.0- Python 3.12
- OpenTelemetry instrumentation enabled
Related issue
#1101 covers a broader subscriber-cleanup lifecycle problem. This report is
limited to the false ERROR span from the intentionally suppressed
remove_sink() KeyError.
Code of Conduct
What happened?
Calling
EventQueueSink.close()more than once is functionally safe, but thesecond close exports an OpenTelemetry
ERRORspan forEventQueueSource.remove_sink.EventQueueSink.close()documents that repeated calls are safe and suppressesKeyErroraroundremove_sink(). However,EventQueueSourceis decoratedwith
@trace_class, soremove_sink()is traced separately. Itsset.remove()raises before the caller suppresses the exception, and thetelemetry wrapper records the exception and sets the span status to
ERROR.This produces a false error observation even though cleanup completes normally.
If the second cleanup happens after the request, the late span can also extend
the apparent trace latency.
Minimal reproduction
With A2A OpenTelemetry instrumentation enabled, the second call records:
The application does not receive the exception because
close()suppresses it.Expected behavior
Repeated or concurrent sink cleanup should not raise an expected exception
inside the traced
remove_sink()method, and noERRORspan should beexported for a successful idempotent close.
Proposed fix
Use
set.discard()when detaching a sink. It preserves removal when the sink ispresent and does nothing when cleanup already removed it.
Environment
a2a-sdk==1.1.0Related issue
#1101 covers a broader subscriber-cleanup lifecycle problem. This report is
limited to the false
ERRORspan from the intentionally suppressedremove_sink()KeyError.Code of Conduct