api: Implement custom events framework in gRPC-Java server by kannanjgithub · Pull Request #12980 · grpc/grpc-java

kannanjgithub · 2026-08-10T09:56:49Z

This adds triggerEvent/onEvent APIs to ServerCall and ServerCall.Listener routing them through ServerStream transport.

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

…pc#12979)

kannanjgithub · 2026-08-10T10:14:09Z

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

…stom-events

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 the source.

sauravzg · 2026-08-31T15:21:50Z

  }

+  @Override
+  public void triggerEvent(Object event) {


do we need a cancellation and close check here? other methods seem to have it.

closeCalled is interesting because it'd require us to make it volatile which may break other assumptions about thread safety.

A check in ServerCallImpl could never prevent concurrent races anyway. Because ServerCall.triggerEvent is intended to be thread-safe, external threads may invoke call.triggerEvent(...) concurrently with the application thread invoking call.close(...).

Even if closeCalled were volatile, thread A could read closeCalled == false a nanosecond before thread B sets closeCalled = true.

An event could therefore always enter stream.triggerEvent(...) while close is in progress.

Thus, the authoritative synchronization point must reside in the transport layer where stream lifecycle state and inbound events are serialize.

The listenerClosed check in AbstractServerStream.triggerEvent completely compensates for not checking closeCalled in ServerCallImpl.triggerEvent, by dropping the event if the listener is closed.

I am less worried about races since we can fix and address them and more worried about the expected behavior.

Right now it seems like we don't check cancellation or closure status here when triggering operations which may be okay if our interface contract is "you should not call after cancellation" . If our contract allows or specifies the behavior after cancellation , I'd assume we check and enforce it here in the implementation.

ServerCall.close's javadoc says

This method implies the caller completed processing the RPC

so that implies that a triggerEvent called after onClose will be ignored.
My earlier reply was concerning races but I agree that we can simply return if callClosed is true when triggerEvent is called, since it provides a fast-path optimization when the same thread invokes call.triggerEvent after invoking call.close().

Unresolving. While looking at it, I realized that closeCalled is non volatile and being used in a thread safe method. I'd guess this is bad behavior?

The thread is using a thread-local variable, it is not even meant for cross thread visibility. So it doesn't fit into the definition you are quoting at all.

We might not be on the same page here? closeCalled is not a thread local. I think we can take this offline and discuss tommorow.

It is not thread-local. What I really meant was that the only scenario where it helps is intra-thread offering a fast path. Where it is visible to other threads (like a volatile), it still does nothing about preventing races.

Let's do this.

Let's try to change the test so that call.close and call.triggerEvent run concurrently on different threads instead of sequentially like they are right now and run a tsan on it. IIUC, that's a very valid behavior that an application can exhibit right? We'll know based on if TSAN fails or not.

public void triggerEvent_afterClose_noop() { call.close(Status.OK, new Metadata()); Object event = new Object(); call.triggerEvent(event); verify(stream, never()).triggerEvent(event); }

TSAN will flag it as a race, I'm not contesting that. But there is a reason TSAN suppressions exist (ex. CL/967859705) for developers to indicate intentionally allowed races that have no adverse impact.

…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

Do we have enough test coverage? I see some of source files with changes but not their corresponding test files?

sauravzg · 2026-09-02T10:27:43Z

  }

+  @Override
+  public void triggerEvent(Object event) {


I am less worried about races since we can fix and address them and more worried about the expected behavior.

Right now it seems like we don't check cancellation or closure status here when triggering operations which may be okay if our interface contract is "you should not call after cancellation" . If our contract allows or specifies the behavior after cancellation , I'd assume we check and enforce it here in the implementation.

…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

kannanjgithub · 2026-09-03T05:40:12Z

Do we have enough test coverage? I see some of source files with changes but not their corresponding test files?

Only the tests for the changes in TransmitStatusRuntimeExceptionInterceptor were missed, added that now.

…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

kannanjgithub added 5 commits August 10, 2026 07:33

Fix forwarding listeners to propagate custom events

e37245c

Address review from server interceptor executor design comments.

2636ebb

Add @experimentalapi annotation to triggerEvent and onEvent (issue gr…

5caf237

…pc#12979)

Add unit test for server stream custom events in AbstractTransportTest

2158b97

kannanjgithub closed this Aug 10, 2026

Fix compilation errors in binder, netty, and okhttp (issue grpc#12979)

605cc03

kannanjgithub reopened this Aug 10, 2026

kannanjgithub requested a review from sauravzg August 10, 2026 12:09

kannanjgithub added 7 commits August 10, 2026 12:52

Implement triggerEvent for Binder transport.

dd549f3

Merge remote-tracking branch 'origin/master' into server-framework-cu…

542f840

…stom-events

sauravzg reviewed Sep 1, 2026

View reviewed changes

kannanjgithub added 2 commits September 2, 2026 06:47

sauravzg reviewed Sep 2, 2026

View reviewed changes

kannanjgithub added 2 commits September 3, 2026 04:47

Uh oh!

Conversation

kannanjgithub commented Aug 10, 2026

Uh oh!

kannanjgithub commented Aug 10, 2026

Uh oh!

sauravzg left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kannanjgithub Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kannanjgithub Sep 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sauravzg left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kannanjgithub commented Sep 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

kannanjgithub Sep 3, 2026 •

edited

Loading

kannanjgithub Sep 4, 2026 •

edited

Loading