diff --git a/api/src/main/java/io/grpc/Contexts.java b/api/src/main/java/io/grpc/Contexts.java
index c62ffc80a38..9c3697dd6d4 100644
--- a/api/src/main/java/io/grpc/Contexts.java
+++ b/api/src/main/java/io/grpc/Contexts.java
@@ -118,6 +118,16 @@ public void onReady() {
         context.detach(previous);
       }
     }
+
+    @Override
+    public void onEvent(Object event) {
+      Context previous = context.attach();
+      try {
+        super.onEvent(event);
+      } finally {
+        context.detach(previous);
+      }
+    }
   }
 
   /**
diff --git a/api/src/main/java/io/grpc/PartialForwardingServerCall.java b/api/src/main/java/io/grpc/PartialForwardingServerCall.java
index a313407b23e..8c8f53cf93c 100644
--- a/api/src/main/java/io/grpc/PartialForwardingServerCall.java
+++ b/api/src/main/java/io/grpc/PartialForwardingServerCall.java
@@ -87,6 +87,11 @@ public SecurityLevel getSecurityLevel() {
     return delegate().getSecurityLevel();
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    delegate().triggerEvent(event);
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("delegate", delegate()).toString();
diff --git a/api/src/main/java/io/grpc/PartialForwardingServerCallListener.java b/api/src/main/java/io/grpc/PartialForwardingServerCallListener.java
index ca2fd0058c9..23e93bb065e 100644
--- a/api/src/main/java/io/grpc/PartialForwardingServerCallListener.java
+++ b/api/src/main/java/io/grpc/PartialForwardingServerCallListener.java
@@ -50,6 +50,11 @@ public void onReady() {
     delegate().onReady();
   }
 
+  @Override
+  public void onEvent(Object event) {
+    delegate().onEvent(event);
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("delegate", delegate()).toString();
diff --git a/api/src/main/java/io/grpc/ServerCall.java b/api/src/main/java/io/grpc/ServerCall.java
index 3db8ac30e83..2e6ee07a23f 100644
--- a/api/src/main/java/io/grpc/ServerCall.java
+++ b/api/src/main/java/io/grpc/ServerCall.java
@@ -100,6 +100,20 @@ public void onComplete() {}
      * <em>another</em> {@code onReady()} callback.
      */
     public void onReady() {}
+
+    /**
+     * A custom event has been triggered by the call.
+     *
+     * <p>This callback is guaranteed to run on the call's executor, serialized with other
+     * 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.
+     */
+    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12979")
+    public void onEvent(Object event) {
+      // Default no-op
+    }
   }
 
   /**
@@ -262,6 +276,20 @@ public String getAuthority() {
     return null;
   }
 
+  /**
+   * Triggers a custom event to be processed by the listener.
+   * The event will be delivered to {@link Listener#onEvent(Object)} on the call's executor.
+   *
+   * <p>This method is thread-safe and can be called from any thread. No events will be delivered
+   * after the RPC is cancelled or completed.
+   *
+   * @param event the event to trigger.
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12979")
+  public void triggerEvent(Object event) {
+    // Default no-op
+  }
+
   /**
    * The {@link MethodDescriptor} for the call.
    */
diff --git a/api/src/test/java/io/grpc/ContextsTest.java b/api/src/test/java/io/grpc/ContextsTest.java
index ec9dc3929a2..974b1aff0c2 100644
--- a/api/src/test/java/io/grpc/ContextsTest.java
+++ b/api/src/test/java/io/grpc/ContextsTest.java
@@ -82,6 +82,11 @@ public void interceptCall_basic() {
         assertSame(uniqueContext, Context.current());
         methodCalls.add(5);
       }
+
+      @Override public void onEvent(Object event) {
+        assertSame(uniqueContext, Context.current());
+        methodCalls.add(6);
+      }
     };
     ServerCall.Listener<Object> wrapped = interceptCall(uniqueContext, call, headers,
         new ServerCallHandler<Object, Object>() {
@@ -101,7 +106,8 @@ public ServerCall.Listener<Object> startCall(
     wrapped.onCancel();
     wrapped.onComplete();
     wrapped.onReady();
-    assertEquals(Arrays.asList(1, 2, 3, 4, 5), methodCalls);
+    wrapped.onEvent(new Object());
+    assertEquals(Arrays.asList(1, 2, 3, 4, 5, 6), methodCalls);
     assertSame(origContext, Context.current());
   }
 
@@ -145,6 +151,10 @@ public void interceptCall_restoresIfListenerThrows() {
       @Override public void onReady() {
         throw new RuntimeException();
       }
+
+      @Override public void onEvent(Object event) {
+        throw new RuntimeException();
+      }
     };
     ServerCall.Listener<Object> wrapped = interceptCall(uniqueContext, call, headers,
         new ServerCallHandler<Object, Object>() {
@@ -180,6 +190,11 @@ public ServerCall.Listener<Object> startCall(
       fail("Exception expected");
     } catch (RuntimeException expected) {
     }
+    try {
+      wrapped.onEvent(new Object());
+      fail("Exception expected");
+    } catch (RuntimeException expected) {
+    }
     assertSame(origContext, Context.current());
   }
 
diff --git a/binder/src/main/java/io/grpc/binder/internal/Inbound.java b/binder/src/main/java/io/grpc/binder/internal/Inbound.java
index 83fc8273d6f..a96f94751d8 100644
--- a/binder/src/main/java/io/grpc/binder/internal/Inbound.java
+++ b/binder/src/main/java/io/grpc/binder/internal/Inbound.java
@@ -668,6 +668,17 @@ protected void deliverCloseAbnormal(Status status) {
       listener.closed(status);
     }
 
+    void triggerEvent(Object event) {
+      synchronized (this) {
+        if (isClosed()) {
+          return;
+        }
+        if (listener != null) {
+          listener.triggerEvent(event);
+        }
+      }
+    }
+
     @GuardedBy("this")
     void onCloseSent(Status status) {
       if (!isClosed()) {
diff --git a/binder/src/main/java/io/grpc/binder/internal/MultiMessageServerStream.java b/binder/src/main/java/io/grpc/binder/internal/MultiMessageServerStream.java
index f54769caefa..7a57138ce22 100644
--- a/binder/src/main/java/io/grpc/binder/internal/MultiMessageServerStream.java
+++ b/binder/src/main/java/io/grpc/binder/internal/MultiMessageServerStream.java
@@ -175,6 +175,11 @@ public void setDecompressor(Decompressor decompressor) {
     // Ignore.
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    inbound.triggerEvent(event);
+  }
+
   @Override
   public void optimizeForDirectExecutor() {
     // Ignore.
diff --git a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java
index ad993b8c93b..3cedce837f6 100644
--- a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java
+++ b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java
@@ -88,6 +88,12 @@ public void onReady() {
     maybeRunPendingSteps();
   }
 
+  @Override
+  public void onEvent(Object event) {
+    pendingSteps.offer(delegate -> delegate.onEvent(event));
+    maybeRunPendingSteps();
+  }
+
   /**
    * Similar to Java8's {@link java.util.function.Consumer}, but redeclared in order to support
    * Android SDK 21.
diff --git a/binder/src/main/java/io/grpc/binder/internal/SingleMessageServerStream.java b/binder/src/main/java/io/grpc/binder/internal/SingleMessageServerStream.java
index 383bd7a2593..5f1dd511f73 100644
--- a/binder/src/main/java/io/grpc/binder/internal/SingleMessageServerStream.java
+++ b/binder/src/main/java/io/grpc/binder/internal/SingleMessageServerStream.java
@@ -167,6 +167,11 @@ public void setDecompressor(Decompressor decompressor) {
     // Ignore.
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    inbound.triggerEvent(event);
+  }
+
   @Override
   public void optimizeForDirectExecutor() {
     // Ignore.
diff --git a/binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java b/binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java
index b0d84f1be74..e556954fd40 100644
--- a/binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java
+++ b/binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java
@@ -289,6 +289,7 @@ public ListenableFuture<Status> checkAuthorizationAsync(int uid) {
     ListenableFuture<Status> authFuture = asyncPolicy.checkAuthorizationAsync(SOME_UID);
     assertThat(awaitResult(settableUid)).isEqualTo(SOME_UID);
     authFuture.cancel(false);
+    executor.submit(() -> {}).get(10, TimeUnit.SECONDS);
 
     assertThat(delegateAuthFuture.isCancelled()).isTrue();
   }
diff --git a/binder/src/test/java/io/grpc/binder/internal/PendingAuthListenerTest.java b/binder/src/test/java/io/grpc/binder/internal/PendingAuthListenerTest.java
index 9cdf123033b..4c17dae302d 100644
--- a/binder/src/test/java/io/grpc/binder/internal/PendingAuthListenerTest.java
+++ b/binder/src/test/java/io/grpc/binder/internal/PendingAuthListenerTest.java
@@ -45,6 +45,7 @@ public void setUp() {
   public void onCallbacks_noOpBeforeStartCall() {
     listener.onReady();
     listener.onMessage("foo");
+    listener.onEvent("bar");
     listener.onHalfClose();
     listener.onComplete();
 
@@ -54,16 +55,19 @@ public void onCallbacks_noOpBeforeStartCall() {
   @Test
   public void onCallbacks_runsPendingCallbacksAfterStartCall() {
     String message = "foo";
+    String event = "bar";
 
     // Act 1
     listener.onReady();
     listener.onMessage(message);
+    listener.onEvent(event);
     listener.startCall(call, headers, next);
 
     // Assert 1
     InOrder order = Mockito.inOrder(delegate);
     order.verify(delegate).onReady();
     order.verify(delegate).onMessage(message);
+    order.verify(delegate).onEvent(event);
 
     // Act 2
     listener.onHalfClose();
diff --git a/core/src/main/java/io/grpc/internal/AbstractServerStream.java b/core/src/main/java/io/grpc/internal/AbstractServerStream.java
index c468cba978a..67dfdc93d42 100644
--- a/core/src/main/java/io/grpc/internal/AbstractServerStream.java
+++ b/core/src/main/java/io/grpc/internal/AbstractServerStream.java
@@ -173,6 +173,16 @@ public final void setListener(ServerStreamListener serverStreamListener) {
     transportState().setListener(serverStreamListener);
   }
 
+  @Override
+  public final void triggerEvent(final Object event) {
+    transportState().runOnTransportThread(new Runnable() {
+      @Override
+      public void run() {
+        transportState().triggerEvent(event);
+      }
+    });
+  }
+
   @Override
   public StatsTraceContext statsTraceContext() {
     return statsTraceCtx;
@@ -259,6 +269,13 @@ public void deframerClosed(boolean hasPartialMessage) {
 
 
 
+    public final void triggerEvent(Object event) {
+      if (listenerClosed) {
+        return;
+      }
+      listener().triggerEvent(event);
+    }
+
     @Override
     protected ServerStreamListener listener() {
       return listener;
diff --git a/core/src/main/java/io/grpc/internal/ServerCallImpl.java b/core/src/main/java/io/grpc/internal/ServerCallImpl.java
index e224384ce8f..1f321d0d8dc 100644
--- a/core/src/main/java/io/grpc/internal/ServerCallImpl.java
+++ b/core/src/main/java/io/grpc/internal/ServerCallImpl.java
@@ -254,6 +254,14 @@ public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
     return method;
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.triggerEvent")) {
+      PerfMark.attachTag(tag);
+      stream.triggerEvent(event);
+    }
+  }
+
   @Override
   public SecurityLevel getSecurityLevel() {
     final Attributes attributes = getAttributes();
@@ -395,5 +403,16 @@ public void onReady() {
         listener.onReady();
       }
     }
+
+    @Override
+    public void triggerEvent(Object event) {
+      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {
+        PerfMark.attachTag(call.tag);
+        if (call.cancelled) {
+          return;
+        }
+        listener.onEvent(event);
+      }
+    }
   }
 }
diff --git a/core/src/main/java/io/grpc/internal/ServerImpl.java b/core/src/main/java/io/grpc/internal/ServerImpl.java
index d9f64c2d473..767d85f443b 100644
--- a/core/src/main/java/io/grpc/internal/ServerImpl.java
+++ b/core/src/main/java/io/grpc/internal/ServerImpl.java
@@ -781,6 +781,9 @@ public void closed(Status status) {}
 
     @Override
     public void onReady() {}
+
+    @Override
+    public void triggerEvent(Object event) {}
   }
 
   /**
@@ -960,6 +963,34 @@ public void runInContext() {
         callExecutor.execute(new OnReady());
       }
     }
+
+    @Override
+    public void triggerEvent(final Object event) {
+      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {
+        PerfMark.attachTag(tag);
+        final Link link = PerfMark.linkOut();
+
+        final class TriggerEvent extends ContextRunnable {
+          TriggerEvent() {
+            super(context);
+          }
+
+          @Override
+          public void runInContext() {
+            try (TaskCloseable ignore = PerfMark.traceTask("ServerCallListener(app).onEvent")) {
+              PerfMark.attachTag(tag);
+              PerfMark.linkIn(link);
+              getListener().triggerEvent(event);
+            } catch (Throwable t) {
+              internalClose(t);
+              throw t;
+            }
+          }
+        }
+
+        callExecutor.execute(new TriggerEvent());
+      }
+    }
   }
 
   @VisibleForTesting
diff --git a/core/src/main/java/io/grpc/internal/ServerStream.java b/core/src/main/java/io/grpc/internal/ServerStream.java
index aa5ba10329c..4c88e94e4d7 100644
--- a/core/src/main/java/io/grpc/internal/ServerStream.java
+++ b/core/src/main/java/io/grpc/internal/ServerStream.java
@@ -87,6 +87,12 @@ public interface ServerStream extends Stream {
    */
   void setListener(ServerStreamListener serverStreamListener);
 
+  /**
+   * Triggers a custom event. Implementations must ensure this is propagated to the
+   * listener on the transport thread.
+   */
+  void triggerEvent(Object event);
+
   /**
    * The context for recording stats and traces for this stream.
    */
diff --git a/core/src/main/java/io/grpc/internal/ServerStreamListener.java b/core/src/main/java/io/grpc/internal/ServerStreamListener.java
index e55217ab422..74de0f2079e 100644
--- a/core/src/main/java/io/grpc/internal/ServerStreamListener.java
+++ b/core/src/main/java/io/grpc/internal/ServerStreamListener.java
@@ -42,4 +42,9 @@ public interface ServerStreamListener extends StreamListener {
    * @param status details about the remote closure
    */
   void closed(Status status);
+
+  /**
+   * Propagates a custom event to the listener. Must be called on the transport thread.
+   */
+  void triggerEvent(Object event);
 }
diff --git a/core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java b/core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java
index 137ba19bfea..5defd17fdd0 100644
--- a/core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java
+++ b/core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java
@@ -361,6 +361,31 @@ public void close_sendTrailersClearsReservedFields() {
     assertEquals("bad", metadataCaptor.getValue().get(InternalStatus.MESSAGE_KEY));
   }
 
+  @Test
+  public void triggerEvent_propagatesToListener() {
+    ServerStreamListener listener = mock(ServerStreamListener.class);
+    stream.transportState().setListener(listener);
+
+    Object event = new Object();
+    stream.triggerEvent(event);
+
+    verify(listener).triggerEvent(event);
+  }
+
+  @Test
+  public void triggerEvent_ignoredAfterClose() {
+    ServerStreamListener listener = mock(ServerStreamListener.class);
+    stream.transportState().setListener(listener);
+
+    stream.close(Status.OK, new Metadata());
+    stream.transportState().complete();
+
+    Object event = new Object();
+    stream.triggerEvent(event);
+
+    verify(listener, never()).triggerEvent(any());
+  }
+
   @Test
   public void changeOnReadyThreshold() {
     stream.setListener(new ServerStreamListenerBase());
@@ -391,6 +416,9 @@ public void halfClosed() {}
 
     @Override
     public void closed(Status status) {}
+
+    @Override
+    public void triggerEvent(Object event) {}
   }
 
   private static class AbstractServerStreamBase extends AbstractServerStream {
diff --git a/core/src/test/java/io/grpc/internal/ServerCallImplTest.java b/core/src/test/java/io/grpc/internal/ServerCallImplTest.java
index 7394c83eab2..4a2de9f3936 100644
--- a/core/src/test/java/io/grpc/internal/ServerCallImplTest.java
+++ b/core/src/test/java/io/grpc/internal/ServerCallImplTest.java
@@ -493,6 +493,32 @@ public void streamListener_unexpectedRuntimeException() {
     assertThat(e).hasMessageThat().isEqualTo("unexpected exception");
   }
 
+  @Test
+  public void triggerEvent_propagatesToStream() {
+    Object event = new Object();
+    call.triggerEvent(event);
+    verify(stream).triggerEvent(event);
+  }
+
+  @Test
+  public void streamListener_triggerEvent() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
+    Object event = new Object();
+    streamListener.triggerEvent(event);
+    verify(callListener).onEvent(event);
+  }
+
+  @Test
+  public void streamListener_triggerEvent_cancelled() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
+    Object event = new Object();
+    streamListener.closed(Status.CANCELLED);
+    streamListener.triggerEvent(event);
+    verify(callListener, never()).onEvent(event);
+  }
+
   private static class LongMarshaller implements Marshaller<Long> {
     @Override
     public InputStream stream(Long value) {
diff --git a/core/src/test/java/io/grpc/internal/ServerImplTest.java b/core/src/test/java/io/grpc/internal/ServerImplTest.java
index 91969dd6910..da2e9646042 100644
--- a/core/src/test/java/io/grpc/internal/ServerImplTest.java
+++ b/core/src/test/java/io/grpc/internal/ServerImplTest.java
@@ -1689,6 +1689,52 @@ public void onReady_runtimeExceptionCancelsCall() {
     }
   }
 
+  @Test
+  public void triggerEvent_delegatesToListener() {
+    JumpToApplicationThreadServerStreamListener listener
+        = new JumpToApplicationThreadServerStreamListener(
+            executor.getScheduledExecutorService(),
+            executor.getScheduledExecutorService(),
+            stream,
+            Context.ROOT.withCancellation(),
+            PerfMark.createTag());
+    ServerStreamListener mockListener = mock(ServerStreamListener.class);
+    listener.setListener(mockListener);
+
+    Object event = new Object();
+    listener.triggerEvent(event);
+
+    verify(mockListener, never()).triggerEvent(any());
+
+    executor.runDueTasks();
+    verify(mockListener).triggerEvent(event);
+  }
+
+  @Test
+  public void triggerEvent_errorCancelsCall() {
+    JumpToApplicationThreadServerStreamListener listener
+        = new JumpToApplicationThreadServerStreamListener(
+            executor.getScheduledExecutorService(),
+            executor.getScheduledExecutorService(),
+            stream,
+            Context.ROOT.withCancellation(),
+            PerfMark.createTag());
+    ServerStreamListener mockListener = mock(ServerStreamListener.class);
+    listener.setListener(mockListener);
+
+    TestError expectedT = new TestError();
+    doThrow(expectedT).when(mockListener).triggerEvent(any());
+
+    listener.triggerEvent(new Object());
+    try {
+      executor.runDueTasks();
+      fail("Expected exception");
+    } catch (TestError t) {
+      assertSame(expectedT, t);
+      ensureServerStateNotLeaked();
+    }
+  }
+
   @Test
   public void binaryLogInstalled() throws Exception {
     final SettableFuture<Boolean> intercepted = SettableFuture.create();
diff --git a/core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java b/core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java
index 5d07de32df9..3898eb8be30 100644
--- a/core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java
+++ b/core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java
@@ -2088,6 +2088,67 @@ public void clientChecksInboundMetadataSize_trailer() throws Exception {
     assertNull(metadata.get(tellTaleKey));
   }
 
+  @Test
+  public void serverStream_triggerEvent() throws Exception {
+    server.start(serverListener);
+    client = newClientTransport(server);
+    startTransport(client, mockClientTransportListener);
+    MockServerTransportListener serverTransportListener
+        = serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    serverTransport = serverTransportListener.transport;
+
+    ClientStream clientStream = client.newStream(
+        methodDescriptor, new Metadata(), callOptions, noopTracers);
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    clientStream.start(clientStreamListener);
+
+    StreamCreation serverStreamCreation
+        = serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    ServerStream serverStream = serverStreamCreation.stream;
+    ServerStreamListenerBase serverStreamListener = serverStreamCreation.listener;
+
+    Object event = new Object();
+    serverStream.triggerEvent(event);
+
+    Object receivedEvent = serverStreamListener.eventQueue.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    assertEquals(event, receivedEvent);
+
+    // Cleanup
+    clientStream.cancel(Status.CANCELLED);
+  }
+
+  @Test
+  public void serverStream_triggerEvent_afterClose() throws Exception {
+    server.start(serverListener);
+    client = newClientTransport(server);
+    startTransport(client, mockClientTransportListener);
+    MockServerTransportListener serverTransportListener
+        = serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    serverTransport = serverTransportListener.transport;
+
+    ClientStream clientStream = client.newStream(
+        methodDescriptor, new Metadata(), callOptions, noopTracers);
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    clientStream.start(clientStreamListener);
+
+    StreamCreation serverStreamCreation
+        = serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    ServerStream serverStream = serverStreamCreation.stream;
+    ServerStreamListenerBase serverStreamListener = serverStreamCreation.listener;
+
+    // Close the stream from client side
+    clientStream.cancel(Status.CANCELLED);
+
+    serverStreamListener.awaitClose(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+
+    Object event = new Object();
+    serverStream.triggerEvent(event);
+
+    // Verify listener did NOT receive the event
+    Object receivedEvent = serverStreamListener.eventQueue.poll(100, TimeUnit.MILLISECONDS);
+    assertNull(receivedEvent);
+  }
+
   /**
    * Helper that simply does an RPC. It can be used similar to a sleep for negative testing: to give
    * time for actions _not_ to happen. Since it is based on doing an actual RPC with actual
diff --git a/core/src/testFixtures/java/io/grpc/internal/ServerStreamListenerBase.java b/core/src/testFixtures/java/io/grpc/internal/ServerStreamListenerBase.java
index aaa70600542..e4ac01912e4 100644
--- a/core/src/testFixtures/java/io/grpc/internal/ServerStreamListenerBase.java
+++ b/core/src/testFixtures/java/io/grpc/internal/ServerStreamListenerBase.java
@@ -89,6 +89,8 @@ public void halfClosed() {
     halfClosedLatch.countDown();
   }
 
+  public final BlockingQueue<Object> eventQueue = new LinkedBlockingQueue<>();
+
   @Override
   public void closed(Status status) {
     if (this.status.isDone()) {
@@ -96,4 +98,12 @@ public void closed(Status status) {
     }
     this.status.set(status);
   }
+
+  @Override
+  public void triggerEvent(Object event) {
+    if (this.status.isDone()) {
+      fail("triggerEvent invoked after closed");
+    }
+    eventQueue.add(event);
+  }
 }
diff --git a/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java b/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java
index a92f10fd5c5..57820b396ad 100644
--- a/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java
+++ b/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java
@@ -429,6 +429,11 @@ public void setListener(ServerStreamListener serverStreamListener) {
         clientStream.setListener(serverStreamListener);
       }
 
+      @Override
+      public void triggerEvent(Object event) {
+        clientStream.triggerServerEvent(event);
+      }
+
       @Override
       public void request(int numMessages) {
         boolean onReady = clientStream.serverRequested(numMessages);
@@ -732,6 +737,20 @@ private synchronized void setListener(ServerStreamListener listener) {
         this.serverStreamListener = listener;
       }
 
+      void triggerServerEvent(final Object event) {
+        synchronized (this) {
+          if (!closed) {
+            syncContext.executeLater(new Runnable() {
+              @Override
+              public void run() {
+                serverStreamListener.triggerEvent(event);
+              }
+            });
+          }
+        }
+        syncContext.drain();
+      }
+
       @Override
       public void request(int numMessages) {
         boolean onReady = serverStream.clientRequested(numMessages);
diff --git a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java b/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java
index ef8d2e5efda..b22be2460f1 100644
--- a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java
+++ b/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java
@@ -1343,6 +1343,10 @@ public void halfClosed() {
     @Override
     public void closed(Status status) {
     }
+
+    @Override
+    public void triggerEvent(Object event) {
+    }
   }
 
   private final class EchoServerListener implements ServerListener {
diff --git a/okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java b/okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java
index 00db6e1d339..d3e50cf3821 100644
--- a/okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java
+++ b/okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java
@@ -1519,6 +1519,10 @@ public void closed(Status status) {
     public void onReady() {
     }
 
+    @Override
+    public void triggerEvent(Object event) {
+    }
+
     static String getContent(InputStream message) throws IOException {
       try {
         return new String(ByteStreams.toByteArray(message), UTF_8);
diff --git a/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryTracingModule.java b/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryTracingModule.java
index 32aab870f0f..088d6dc9845 100644
--- a/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryTracingModule.java
+++ b/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryTracingModule.java
@@ -452,6 +452,13 @@ public void onReady() {
         delegate().onReady();
       }
     }
+
+    @Override
+    public void onEvent(Object event) {
+      try (Scope scope = context.makeCurrent()) {
+        delegate().onEvent(event);
+      }
+    }
   }
 
   @VisibleForTesting
diff --git a/opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java b/opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java
index 0b5bff1d036..ee7e86e05cc 100644
--- a/opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java
+++ b/opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java
@@ -948,6 +948,11 @@ public void onCancel() {
       public void onComplete() {
         callbackSpan.set(Span.fromContext(Context.current()));
       }
+
+      @Override
+      public void onEvent(Object event) {
+        callbackSpan.set(Span.fromContext(Context.current()));
+      }
     };
     ServerInterceptor interceptor = tracingModule.getServerSpanPropagationInterceptor();
     @SuppressWarnings("unchecked")
@@ -967,6 +972,8 @@ public void onComplete() {
     assertEquals(callbackSpan.get(), Span.getInvalid());
     listener.onComplete();
     assertEquals(callbackSpan.get(), Span.getInvalid());
+    listener.onEvent(new Object());
+    assertEquals(callbackSpan.get(), Span.getInvalid());
 
     Span parentSpan = tracerRule.spanBuilder("parent-span").startSpan();
     io.grpc.Context context = io.grpc.Context.current().withValue(
@@ -990,6 +997,9 @@ public void onComplete() {
       listener.onComplete();
       assertEquals(callbackSpan.get().getSpanContext().getTraceId(),
           parentSpan.getSpanContext().getTraceId());
+      listener.onEvent(new Object());
+      assertEquals(callbackSpan.get().getSpanContext().getTraceId(),
+          parentSpan.getSpanContext().getTraceId());
     } finally {
       context.detach(previous);
     }
diff --git a/util/src/main/java/io/grpc/util/TransmitStatusRuntimeExceptionInterceptor.java b/util/src/main/java/io/grpc/util/TransmitStatusRuntimeExceptionInterceptor.java
index b477ae1fdfb..e4b364bd532 100644
--- a/util/src/main/java/io/grpc/util/TransmitStatusRuntimeExceptionInterceptor.java
+++ b/util/src/main/java/io/grpc/util/TransmitStatusRuntimeExceptionInterceptor.java
@@ -104,6 +104,15 @@ public void onReady() {
         }
       }
 
+      @Override
+      public void onEvent(Object event) {
+        try {
+          super.onEvent(event);
+        } catch (StatusRuntimeException e) {
+          closeWithException(e);
+        }
+      }
+
       private void closeWithException(StatusRuntimeException t) {
         Metadata metadata = t.getTrailers();
         if (metadata == null) {
@@ -276,5 +285,15 @@ public void run() {
         throw new RuntimeException(ERROR_MSG, e);
       }
     }
+
+    @Override
+    public void triggerEvent(final Object event) {
+      serializingExecutor.execute(new Runnable() {
+        @Override
+        public void run() {
+          SerializingServerCall.super.triggerEvent(event);
+        }
+      });
+    }
   }
 }
diff --git a/util/src/test/java/io/grpc/util/UtilServerInterceptorsTest.java b/util/src/test/java/io/grpc/util/UtilServerInterceptorsTest.java
index a4691d8bdec..4cc229c1851 100644
--- a/util/src/test/java/io/grpc/util/UtilServerInterceptorsTest.java
+++ b/util/src/test/java/io/grpc/util/UtilServerInterceptorsTest.java
@@ -104,6 +104,11 @@ public void onComplete() {
       public void onReady() {
         throw exception;
       }
+
+      @Override
+      public void onEvent(Object event) {
+        throw exception;
+      }
     };
 
     ServerServiceDefinition intercepted = ServerInterceptors.intercept(
@@ -116,7 +121,36 @@ public void onReady() {
     getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onComplete();
     getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onHalfClose();
     getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onReady();
-    assertEquals(5, call.numCloses);
+    getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers)
+        .onEvent(new Object());
+    assertEquals(6, call.numCloses);
+  }
+
+  @Test
+  public void statusRuntimeExceptionTransmitter_serializingServerCall_triggerEvent() {
+    final java.util.concurrent.atomic.AtomicReference<Object> eventRef =
+        new java.util.concurrent.atomic.AtomicReference<>();
+    FakeServerCall<Void, Void> call = new FakeServerCall<Void, Void>(Status.OK, new Metadata()) {
+      @Override
+      public void triggerEvent(Object event) {
+        eventRef.set(event);
+      }
+    };
+    final java.util.concurrent.atomic.AtomicReference<ServerCall<Void, Void>> interceptedCall =
+        new java.util.concurrent.atomic.AtomicReference<>();
+    listener = new VoidCallListener() {
+      @Override
+      public void onCall(ServerCall<Void, Void> call, Metadata headers) {
+        interceptedCall.set(call);
+      }
+    };
+    ServerServiceDefinition intercepted = ServerInterceptors.intercept(
+        serviceDefinition,
+        Arrays.asList(TransmitStatusRuntimeExceptionInterceptor.instance()));
+    getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers);
+    Object testEvent = new Object();
+    interceptedCall.get().triggerEvent(testEvent);
+    assertEquals(testEvent, eventRef.get());
   }
 
   @Test