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: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.

The expected behavior should be documented on the interface on if it's callable after cancellation/completion , if yes, what's the expected behavior.

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.

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.

and why did we need it now and not before this PR? or is this an unrelated change?

…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:23:06Z

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


and why did we need it now and not before this PR? or is this an unrelated change?

sauravzg · 2026-09-02T10:24:06Z

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


The expected behavior should be documented on the interface on if it's callable after cancellation/completion , if yes, what's the expected behavior.

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.

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

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!

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!

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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants