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:15:32Z

+     * callbacks (like {@link #onMessage}, {@link #onHalfClose}). This means the implementation
+     * does not need internal synchronization to access call-specific state.
+     *
+     * @param event the triggered event.


Should we clarify what we expect event to be? I'd assume we want a runnable? Object seems vague, what's the behavior when called with an arbitrary object type? Also, stuff about whether the operations should be blocking /non-blocking etc.

It is not expected to be a runnable, but just a custom Event object, just like in the case of onMessage. Runnable and task executor handling is done by the server framework.
About blocking/non-blocking etc, the javadoc for the interface applies.

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

+     * @param event the triggered event.
+     */
+    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12979")
+    public void onEvent(Object event) {


Any invariants on this ? Can this be called after cancellation(I assume no)?

The framework code (ServerStreamListenerImpl) calls onEvent and it will drop it if the call is cancelled.

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.

sauravzg · 2026-09-01T09:46:20Z

+
+    @Override
+    public void triggerEvent(Object event) {
+      if (call.cancelled) {


similar to above about checking close. I don't know the solution however. Is it as simple as making it volatile or are there other invariants involved here?

call.cancelled is volatile. It indicates abnormal RPC cancellation (from the client, network disconnect, deadline expiration, or context cancellation), and is set by the cancellation context listener in ServerStreamListenerImpl that runs on a dedicated cancellation executor.

sauravzg · 2026-09-01T10:01:36Z

We may need to audit the existing implementations that may need to be updated with some implementations. I see that we've updated the PartialForwarding... .

But I was able to spot a few ones that seemed like they would not function as expected without it like OpenTelemetryTracingModule and TransmitStatusRuntimeExceptionInterceptor

Maybe we should do an audit of existing implementations to identify what needs to be updated and what doesn't? I am slightly worried about some arbitrary filter or interceptor that uses the default implementation reducing the entire chain to no-op.

On the binder side, we may have PendingAuth...Listener which might need this as well.

Summary of Audit Findings

Implementation Type Current Behavior for triggerEvent / onEvent Issue / Risk Action Required

PendingAuthListener (binder) Direct ServerCall.Listener Does not implement onEvent High: Silently drops / no-ops all custom events while auth is pending (and even after auth completes). Must fix: Implement onEvent to buffer and replay to delegate.

TransmitStatusRuntimeExceptionInterceptor (util) SimpleForwardingServerCallListener & SerializingServerCall Neither overrides onEvent nor triggerEvent Medium: (1) Application onEvent throwing StatusRuntimeException is not converted to call.close(). (2) triggerEvent bypasses SerializingExecutor. Should fix: Wrap onEvent in try/catch; serialize triggerEvent on executor.

OpenTelemetryTracingModule (opentelemetry) SimpleForwardingServerCallListener Does not override onEvent Low/Medium: Forwards event, but runs without the OTel trace Scope / Span active. Should fix: Override onEvent with try (Scope scope = context.makeCurrent()).

Contexts.java (api) SimpleForwardingServerCallListener Overrides onEvent to attach/detach context None (Already updated on our branch). None.

ServerInterceptors.wrapHandler (api) PartialForwardingServerCall / Listener Inherits from base forwarding classes None (Properly delegates). None.

StreamingServerCallListener (stub) Direct ServerCall.Listener Inherits default empty onEvent Design limitation: Events currently terminate at the stub layer unless intercepted upstream. Stub-layer support (e.g. ServerCallStreamObserver.setOnEventHandler) needed if stubs need events, will address this separately.

Standard Interceptors (HeaderServerInterceptor, OrcaMetricReportingServerInterceptor, InternalLoggingServerInterceptor, BinlogHelper, MetadataExchanger) SimpleForwarding* classes Inherit default delegation from PartialForwarding* None (Automatically delegate). None.

Rejection Listeners (RbacFilter, XdsServerWrapper, etc.) Direct new ServerCall.Listener(){} Empty default implementation None (Intentional no-op: call was rejected/closed immediately). None.

I have now implemented the custom Event handling in PendingAuthListener, TransmitStatusRuntimeExceptionInterceptor and OpenTelemetryTracingModule.

sauravzg · 2026-09-01T10:56:11Z

  }

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


Some of the other implementations have perfmark tags attached and some other things. Do we need that here as well?

I have added permark tags now.

sauravzg · 2026-09-01T10:57:38Z

    ListenableFuture<Status> authFuture = asyncPolicy.checkAuthorizationAsync(SOME_UID);
    assertThat(awaitResult(settableUid)).isEqualTo(SOME_UID);
    authFuture.cancel(false);
+    executor.submit(() -> {}).get(10, TimeUnit.SECONDS);


what are we doing here? Seems like no-op to me.

It is causing a wait for the cancellation task submitted to the executor to be complete before the assertion in the next statement.

sauravzg · 2026-09-01T12:25:43Z

+        }
+        localListener = listener;
+      }
+      if (localListener != null) {


This seems to be excluded from the synchronized block. Doesn't this create an issue around thread safety for the localListener which may not be thread safe?

Yes, but this has already been fixed in commit e832422.

…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

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

Implementation	Type	Current Behavior for `triggerEvent` / `onEvent`	Issue / Risk	Action Required
`PendingAuthListener` (`binder`)	Direct `ServerCall.Listener`	Does not implement `onEvent`	High: Silently drops / no-ops all custom events while auth is pending (and even after auth completes).	Must fix: Implement `onEvent` to buffer and replay to delegate.
`TransmitStatusRuntimeExceptionInterceptor` (`util`)	`SimpleForwardingServerCallListener` & `SerializingServerCall`	Neither overrides `onEvent` nor `triggerEvent`	Medium: (1) Application `onEvent` throwing `StatusRuntimeException` is not converted to `call.close()`. (2) `triggerEvent` bypasses `SerializingExecutor`.	Should fix: Wrap `onEvent` in try/catch; serialize `triggerEvent` on executor.
`OpenTelemetryTracingModule` (`opentelemetry`)	`SimpleForwardingServerCallListener`	Does not override `onEvent`	Low/Medium: Forwards event, but runs without the OTel trace `Scope` / `Span` active.	Should fix: Override `onEvent` with `try (Scope scope = context.makeCurrent())`.
`Contexts.java` (`api`)	`SimpleForwardingServerCallListener`	Overrides `onEvent` to attach/detach context	None (Already updated on our branch).	None.
`ServerInterceptors.wrapHandler` (`api`)	`PartialForwardingServerCall` / `Listener`	Inherits from base forwarding classes	None (Properly delegates).	None.
`StreamingServerCallListener` (`stub`)	Direct `ServerCall.Listener`	Inherits default empty `onEvent`	Design limitation: Events currently terminate at the stub layer unless intercepted upstream.	Stub-layer support (e.g. `ServerCallStreamObserver.setOnEventHandler`) needed if stubs need events, will address this separately.
Standard Interceptors (`HeaderServerInterceptor`, `OrcaMetricReportingServerInterceptor`, `InternalLoggingServerInterceptor`, `BinlogHelper`, `MetadataExchanger`)	`SimpleForwarding*` classes	Inherit default delegation from `PartialForwarding*`	None (Automatically delegate).	None.
Rejection Listeners (`RbacFilter`, `XdsServerWrapper`, etc.)	Direct `new ServerCall.Listener(){}`	Empty default implementation	None (Intentional no-op: call was rejected/closed immediately).	None.

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!

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!

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

Summary of Audit Findings

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants