api: Implement custom events framework in gRPC-Java server - #12980
Merged
kannanjgithub merged 21 commits intoSep 10, 2026
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
Consistent with ServerCallImpl.request(int), do not short-circuit on the non-volatile closeCalled boolean in triggerEvent. This ensures ServerCallImpl delegates triggerEvent to the underlying ServerStream, where stream lifecycle state and serialization are authoritatively managed in the transport layer. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
sauravzg
approved these changes
Sep 8, 2026
…ner.onEvent Add deadlock avoidance notes to the Javadoc of ServerCall.triggerEvent() and ServerCall.Listener.onEvent(). Transports such as Binder may hold transport locks (e.g. Inbound.this) while synchronously dispatching triggerEvent and onEvent. If application or interceptor code holds internal locks when calling ServerCall methods or acquires them inside onEvent(), lock order inversion deadlocks can occur. TAG=agy CONV=b39523be-9572-456e-a428-8b0e255067da
kannanjgithub
force-pushed
the
server-framework-custom-events
branch
from
September 9, 2026 07:04
2df5f45 to
ab458cc
Compare
…tests
In ExternalProcessorClientInterceptorTest, tests in Category 27
(clientInterceptor_contextPropagated*) configure dedicated single-thread
executors for thread-boundary context propagation testing and register
the in-process channels and servers with GrpcCleanupRule.
Previously, the tests exited their try-blocks after observing only
intermediate latches (such as downstream start or ext-proc call start),
then immediately triggered call cancellation and entered finally-block
executor shutdown. When custom executors were shut down while stream
cancellation or completion tasks were still in flight or being queued,
in-process transport streams could not cleanly complete. Consequently,
GrpcCleanupRule.after() timed out awaiting channel/server termination,
failing with:
java.lang.AssertionError: Resources could not be released in time
Fix this by awaiting completion/closure latches on both the client call
listener (onClose) and the mock external processor server (onError /
onCompleted) before exiting the try block and shutting down the executors.
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.Addresses #7868 for server interceptors.