api: Implement custom events framework in gRPC-Java server - #12980
Open
kannanjgithub wants to merge 18 commits into
Open
api: Implement custom events framework in gRPC-Java server#12980kannanjgithub wants to merge 18 commits into
kannanjgithub wants to merge 18 commits into
Conversation
This adds triggerEvent/onEvent APIs to ServerCall and ServerCall.Listener, routing them through ServerStream transport to ensure thread-safety (especially for SerializeReentrantCallsDirectExecutor). TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
Contributor
Author
|
Need to implement methods in Binder transport. |
…framework. - Added unit tests in AbstractServerStreamTest for triggerEvent propagation and close behavior. - Updated ContextsTest to cover onEvent propagation in ContextualizedServerCallListener. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
Synchronized with the executor before asserting cancellation of the delegate future to ensure that transformAsync has finished processing the delegate future and propagated the cancellation. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
… behavior. - Added unit tests in ServerImplTest for JumpToApplicationThreadServerStreamListener.triggerEvent. - Added serverStream_triggerEvent_afterClose in AbstractTransportTest to verify events are ignored after stream closure. - Updated Inbound.ServerInbound to check isClosed() before triggering events. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
Wait for the server stream to be fully closed (via awaitClose) before calling triggerEvent, to ensure the transport has processed the cancellation and marked the listener as closed. This fixes flakiness in slower transports like Jetty. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
Updated ServerInbound.triggerEvent to invoke the listener's triggerEvent callback inside the synchronized(this) block. This ensures that the check for isClosed() and the invocation of the listener are atomic relative to stream closure (which also runs under the same lock). This prevents a race where triggerEvent could be called on the listener after the stream has been closed, which would result in out-of-order events delivered to the application. This is consistent with how other listener callbacks (like closed and halfClosed) are delivered in Inbound.java. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
sauravzg
reviewed
Sep 1, 2026
…ExceptionInterceptor, and OpenTelemetryTracingModule. - binder: Implement onEvent in PendingAuthListener to buffer and replay custom events to the delegate once auth completes, preventing events from being dropped. - util: Handle onEvent in TransmitStatusRuntimeExceptionInterceptor listener wrapper to catch StatusRuntimeException and close the call. Serialize triggerEvent on SerializingServerCall's executor. - opentelemetry: Implement onEvent in ContextServerCallListener to attach OpenTelemetry trace context and scope during delegate invocation. - Add unit tests for all updated implementations. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…mListenerImpl triggerEvent Wrap ServerCallImpl.triggerEvent and ServerStreamListenerImpl.triggerEvent in PerfMark.traceTask with PerfMark.attachTag, aligning them with sendMessage, sendHeaders, close, request, and listener callbacks. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
sauravzg
reviewed
Sep 2, 2026
sauravzg
left a comment
sauravzg
left a comment
Contributor
There was a problem hiding this comment.
Do we have enough test coverage? I see some of source files with changes but not their corresponding test files?
…d is true Check closeCalled before dispatching triggerEvent to the transport stream, avoiding unnecessary task allocations and transport hops if the call has already been closed. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…stom event changes - Test onEvent throwing StatusRuntimeException closes the call with status and trailers. - Test onEvent throwing StatusRuntimeException on an already closed call does not trigger duplicate close. - Test SerializingServerCall executes triggerEvent sequentially in FIFO order on serializingExecutor. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
Contributor
Author
Only the tests for the changes in |
…e stream cleanup in AbstractTransportTest
In MockServerTransportListener.streamCreated(), stream.setListener(listener)
was called after streams.add(StreamCreation(...)). This created a race
condition where a test thread calling takeStreamOrFail() could dequeue
the stream and call serverStream.triggerEvent() before stream.setListener()
was called by the transport/container thread. When this occurred (e.g. in
TomcatTransportTest on multi-core runners), ServletServerStream invoked
transportState.triggerEvent() on the test thread, saw a null listener,
threw a NullPointerException (swallowed by SerializingExecutor), and
never enqueued the event into the listener queue, leading to a timeout
and assertion failure:
expected:<...Object@...> but was:<null>
Setting stream.setListener(listener) before enqueuing to streams guarantees
that any thread consuming the StreamCreation will always observe a fully
initialized listener.
Additionally, in AbstractTransportTest.serverStream_triggerEvent(), replace
clientStream.cancel(Status.CANCELLED) with serverStream.close(Status.OK, ...)
for clean stream closure instead of leaving an uncoordinated client RST_STREAM
in flight during container tearDown.
TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This adds triggerEvent/onEvent APIs to
ServerCallandServerCall.Listenerrouting them throughServerStreamtransport.