Skip to content
Open
Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
9d14356
outlier-detection: exclude client and hedging cancellations from call…
AgraVator Jul 22, 2026
7abb485
Update @since tag to 1.84.0 and add tracer cancellation in FailingCli…
AgraVator Jul 22, 2026
f9cc87f
core, inprocess, binder: synchronize client cancellation tracer notif…
AgraVator Jul 23, 2026
120e35c
test: add unit tests for race condition and transport tracer cancella…
AgraVator Jul 24, 2026
f9dd0fb
test: add inprocess client stream cancellation unit tests
AgraVator Jul 24, 2026
fe581ae
Revert "test: add inprocess client stream cancellation unit tests"
AgraVator Jul 24, 2026
5e558a5
test: fix checkstyle line length violations in InProcessTransportTest
AgraVator Jul 24, 2026
634be81
Revert "test: fix checkstyle line length violations in InProcessTrans…
AgraVator Jul 24, 2026
20682d9
test: fix checkstyle violations and missing server.start in InProcess…
AgraVator Jul 27, 2026
bf90eb5
util: exclude client cancellations and deadline exceeded from outlier…
AgraVator Aug 3, 2026
6d3cf19
core: trigger clientCancelled in closeListener based on stopDelivery …
AgraVator Aug 3, 2026
9331806
util: restore incrementCallCount(boolean success) method signature
AgraVator Aug 3, 2026
5f3af3d
test: add explicit stopDelivery unit tests in AbstractClientStreamTest
AgraVator Aug 3, 2026
be50f7b
test: add comprehensive stopDelivery and server vs client cancellatio…
AgraVator Aug 3, 2026
c6e2f4a
test: remove redundant directAssertions test case
AgraVator Aug 3, 2026
f27ce70
core, binder: notify clientCancelled directly in transportReportStatu…
AgraVator Aug 5, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions api/src/main/java/io/grpc/ClientStreamTracer.java
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,15 @@ public void inboundTrailers(Metadata trailers) {
public void addOptionalLabel(String key, String value) {
}

/**
* The stream was cancelled from the client side before a normal response was received.
*
* @param status the cancellation status
* @since 1.84.0
*/
public void cancelled(Status status) {
}

/**
* Factory class for {@link ClientStreamTracer}.
*/
Expand Down
3 changes: 3 additions & 0 deletions binder/src/main/java/io/grpc/binder/internal/Inbound.java
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,9 @@ private final void deliverInternal() {

@GuardedBy("this")
final void closeOnCancel(Status status) {
if (!isClosed() && statsTraceContext != null) {
Comment thread
AgraVator marked this conversation as resolved.
statsTraceContext.clientCancelled(status);
}
closeAbnormal(Status.CANCELLED, status, false);
}

Expand Down
3 changes: 3 additions & 0 deletions core/src/main/java/io/grpc/internal/AbstractClientStream.java
Original file line number Diff line number Diff line change
Expand Up @@ -457,6 +457,9 @@ private void closeListener(
Status status, RpcProgress rpcProgress, Metadata trailers) {
if (!listenerClosed) {
listenerClosed = true;
if (status.getCode() == Status.Code.CANCELLED) {
Comment thread
AgraVator marked this conversation as resolved.
Outdated
statsTraceCtx.clientCancelled(status);
}
statsTraceCtx.streamClosed(status);
if (getTransportTracer() != null) {
getTransportTracer().reportStreamClosed(status.isOk());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,11 @@ public void addOptionalLabel(String key, String value) {
delegate().addOptionalLabel(key, value);
}

@Override
public void cancelled(Status status) {
delegate().cancelled(status);
}

@Override
public void streamClosed(Status status) {
delegate().streamClosed(status);
Expand Down
13 changes: 13 additions & 0 deletions core/src/main/java/io/grpc/internal/StatsTraceContext.java
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,19 @@ public void serverCallMethodResolved(MethodDescriptor<?, ?> method) {
}
}

/**
* See {@link ClientStreamTracer#cancelled}. For client-side only.
*
* <p>Called from abstract stream implementations.
*/
public void clientCancelled(Status status) {
for (StreamTracer tracer : tracers) {
if (tracer instanceof ClientStreamTracer) {
((ClientStreamTracer) tracer).cancelled(status);
}
}
}

/**
* See {@link StreamTracer#streamClosed}. This may be called multiple times, and only the first
* value will be taken.
Expand Down
42 changes: 42 additions & 0 deletions core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ClientStreamTracer;
import io.grpc.Codec;
import io.grpc.Deadline;
import io.grpc.Grpc;
Expand Down Expand Up @@ -155,6 +156,47 @@ public void cancel(Status errorStatus) {
verify(mockListener).closed(any(Status.class), same(PROCESSED), any(Metadata.class));
}

@Test
public void cancel_notifiesStatsTraceContext() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {
@Override
public void cancel(Status errorStatus) {
state.transportReportStatus(errorStatus, true, new Metadata());
}
}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

Status cancelStatus = Status.CANCELLED.withDescription("Cancelled by test");
stream.cancel(cancelStatus);

verify(mockTracer).cancelled(cancelStatus);
}

@Test
public void transportReportStatus_okFirst_lateCancellationDoesNotNotifyTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {
@Override
public void cancel(Status errorStatus) {
state.transportReportStatus(errorStatus, true, new Metadata());
}
}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Report Status.OK first
state.transportReportStatus(Status.OK, false, new Metadata());

// Subsequent late cancellation
stream.cancel(Status.CANCELLED.withDescription("Late cancel"));

verify(mockTracer, never()).cancelled(any(Status.class));
}

@Test
public void startFailsOnNullListener() {
AbstractClientStream stream =
Expand Down
49 changes: 49 additions & 0 deletions core/src/test/java/io/grpc/internal/StatsTraceContextTest.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
/*
* Copyright 2026 The gRPC Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package io.grpc.internal;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;

import io.grpc.ClientStreamTracer;
import io.grpc.ServerStreamTracer;
import io.grpc.Status;
import io.grpc.StreamTracer;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

/** Unit tests for {@link StatsTraceContext}. */
@RunWith(JUnit4.class)
public class StatsTraceContextTest {

@Test
public void clientCancelled_notifiesClientStreamTracers() {
ClientStreamTracer clientTracer = mock(ClientStreamTracer.class);
ServerStreamTracer serverTracer = mock(ServerStreamTracer.class);

StatsTraceContext statsTraceCtx = new StatsTraceContext(
new StreamTracer[] {clientTracer, serverTracer});

Status cancelledStatus = Status.CANCELLED.withDescription("Client cancelled");
statsTraceCtx.clientCancelled(cancelledStatus);

verify(clientTracer).cancelled(cancelledStatus);
verifyNoInteractions(serverTracer);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -846,6 +846,7 @@ public void cancel(Status reason) {
if (!internalCancel(serverStatus, serverStatus)) {
return;
}
statsTraceCtx.clientCancelled(reason);
serverStream.clientCancelled(reason);
streamClosed();
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

import io.grpc.CallOptions;
import io.grpc.ClientCall;
import io.grpc.ClientStreamTracer;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
Expand Down Expand Up @@ -236,6 +237,56 @@ public void basicStreamInProcess() throws Exception {
serverStream.close(status, new Metadata());
}

@Test
public void clientStream_cancel_notifiesTracerCancelled() throws Exception {
server = newServer(Arrays.asList(serverStreamTracerFactory));
server.start(serverListener);
client = newClientTransport(server);
startTransport(client, mockClientTransportListener);
MockServerTransportListener serverTransportListener =
serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
serverTransport = serverTransportListener.transport;

ClientStreamTracer mockTracer = org.mockito.Mockito.mock(ClientStreamTracer.class);
ClientStream clientStream = client.newStream(
methodDescriptor, new Metadata(), CallOptions.DEFAULT,
new ClientStreamTracer[] {mockTracer});
ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
clientStream.start(clientStreamListener);

Status cancelStatus = Status.CANCELLED.withDescription("Client cancelled");
clientStream.cancel(cancelStatus);

org.mockito.Mockito.verify(mockTracer).cancelled(cancelStatus);
}

@Test
public void clientStream_cancelAfterServerClose_doesNotNotifyTracerCancelled() throws Exception {
server = newServer(Arrays.asList(serverStreamTracerFactory));
server.start(serverListener);
client = newClientTransport(server);
startTransport(client, mockClientTransportListener);
MockServerTransportListener serverTransportListener =
serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
serverTransport = serverTransportListener.transport;

ClientStreamTracer mockTracer = org.mockito.Mockito.mock(ClientStreamTracer.class);
ClientStream clientStream = client.newStream(
methodDescriptor, new Metadata(), CallOptions.DEFAULT,
new ClientStreamTracer[] {mockTracer});
ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
clientStream.start(clientStreamListener);
StreamCreation serverStreamCreation =
serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
ServerStream serverStream = serverStreamCreation.stream;

serverStream.close(Status.OK, new Metadata());
clientStream.cancel(Status.CANCELLED.withDescription("Late cancellation"));

org.mockito.Mockito.verify(mockTracer, org.mockito.Mockito.never())
.cancelled(org.mockito.Mockito.any(Status.class));
}

private void assertAssumedMessageSize(
TestStreamTracer streamTracerSender, TestStreamTracer streamTracerReceiver) {
if (isEnabledSupportTracingMessageSizes()) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,11 @@ public void addOptionalLabel(String key, String value) {
delegate().addOptionalLabel(key, value);
}

@Override
public void cancelled(Status status) {
delegate().cancelled(status);
}

@Override
public void streamClosed(Status status) {
delegate().streamClosed(status);
Expand Down
23 changes: 21 additions & 2 deletions util/src/main/java/io/grpc/util/OutlierDetectionLoadBalancer.java
Original file line number Diff line number Diff line change
Expand Up @@ -477,22 +477,41 @@ public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata header
if (delegateFactory != null) {
ClientStreamTracer delegateTracer = delegateFactory.newClientStreamTracer(info, headers);
return new ForwardingClientStreamTracer() {
private volatile boolean cancelled;

@Override
protected ClientStreamTracer delegate() {
return delegateTracer;
}

@Override
public void cancelled(Status status) {
cancelled = true;
delegate().cancelled(status);
}

@Override
public void streamClosed(Status status) {
tracker.incrementCallCount(status.isOk());
if (!cancelled) {
tracker.incrementCallCount(status.isOk());
}
delegate().streamClosed(status);
}
};
} else {
return new ClientStreamTracer() {
private volatile boolean cancelled;

@Override
public void cancelled(Status status) {
cancelled = true;
}

@Override
public void streamClosed(Status status) {
tracker.incrementCallCount(status.isOk());
if (!cancelled) {
tracker.incrementCallCount(status.isOk());
}
}
};
}
Expand Down
Loading
Loading