api: Implement custom events framework in gRPC-Java server - #12980
api: Implement custom events framework in gRPC-Java server#12980kannanjgithub wants to merge 13 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
|
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
| * 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. |
There was a problem hiding this comment.
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.
| * @param event the triggered event. | ||
| */ | ||
| @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12979") | ||
| public void onEvent(Object event) { |
There was a problem hiding this comment.
Any invariants on this ? Can this be called after cancellation(I assume no)?
| } | ||
|
|
||
| @Override | ||
| public void triggerEvent(Object event) { |
There was a problem hiding this comment.
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.
|
|
||
| @Override | ||
| public void triggerEvent(Object event) { | ||
| if (call.cancelled) { |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
On the binder side, we may have PendingAuth...Listener which might need this as well.
| } | ||
|
|
||
| @Override | ||
| public void triggerEvent(Object event) { |
There was a problem hiding this comment.
Some of the other implementations have perfmark tags attached and some other things. Do we need that here as well?
| ListenableFuture<Status> authFuture = asyncPolicy.checkAuthorizationAsync(SOME_UID); | ||
| assertThat(awaitResult(settableUid)).isEqualTo(SOME_UID); | ||
| authFuture.cancel(false); | ||
| executor.submit(() -> {}).get(10, TimeUnit.SECONDS); |
There was a problem hiding this comment.
what are we doing here? Seems like no-op to me.
| } | ||
| localListener = listener; | ||
| } | ||
| if (localListener != null) { |
There was a problem hiding this comment.
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?
2 participants
This adds triggerEvent/onEvent APIs to
ServerCallandServerCall.Listenerrouting them throughServerStreamtransport.