diff --git a/api/src/main/java/io/grpc/ClientStreamTracer.java b/api/src/main/java/io/grpc/ClientStreamTracer.java index 8e11e781e7c..537457d2a9e 100644 --- a/api/src/main/java/io/grpc/ClientStreamTracer.java +++ b/api/src/main/java/io/grpc/ClientStreamTracer.java @@ -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}. */ 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..7eea5ee78a1 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Inbound.java +++ b/binder/src/main/java/io/grpc/binder/internal/Inbound.java @@ -268,6 +268,9 @@ private final void deliverInternal() { @GuardedBy("this") final void closeOnCancel(Status status) { + if (!isClosed() && statsTraceContext != null) { + statsTraceContext.clientCancelled(status); + } closeAbnormal(Status.CANCELLED, status, false); } diff --git a/core/src/main/java/io/grpc/internal/AbstractClientStream.java b/core/src/main/java/io/grpc/internal/AbstractClientStream.java index bce1820b482..2e5b333bc36 100644 --- a/core/src/main/java/io/grpc/internal/AbstractClientStream.java +++ b/core/src/main/java/io/grpc/internal/AbstractClientStream.java @@ -197,6 +197,9 @@ public final void halfClose() { @Override public final void cancel(Status reason) { Preconditions.checkArgument(!reason.isOk(), "Should not cancel with OK status"); + if (cancelled || transportState().isListenerClosed()) { + return; + } cancelled = true; abstractClientStreamSink().cancel(reason); } @@ -251,6 +254,10 @@ protected TransportState( } } + protected final boolean isListenerClosed() { + return listenerClosed; + } + private void setFullStreamDecompression(boolean fullStreamDecompression) { this.fullStreamDecompression = fullStreamDecompression; } @@ -435,13 +442,13 @@ public final void transportReportStatus( if (deframerClosed) { deframerClosedTask = null; - closeListener(status, rpcProgress, trailers); + closeListener(status, rpcProgress, trailers, stopDelivery); } else { deframerClosedTask = new Runnable() { @Override public void run() { - closeListener(status, rpcProgress, trailers); + closeListener(status, rpcProgress, trailers, stopDelivery); } }; closeDeframer(stopDelivery); @@ -454,9 +461,12 @@ public void run() { * @throws IllegalStateException if the call has not yet been started. */ private void closeListener( - Status status, RpcProgress rpcProgress, Metadata trailers) { + Status status, RpcProgress rpcProgress, Metadata trailers, boolean stopDelivery) { if (!listenerClosed) { listenerClosed = true; + if (stopDelivery) { + statsTraceCtx.clientCancelled(status); + } statsTraceCtx.streamClosed(status); if (getTransportTracer() != null) { getTransportTracer().reportStreamClosed(status.isOk()); diff --git a/core/src/main/java/io/grpc/internal/ForwardingClientStreamTracer.java b/core/src/main/java/io/grpc/internal/ForwardingClientStreamTracer.java index e7679ea14cc..6dc18fdf627 100644 --- a/core/src/main/java/io/grpc/internal/ForwardingClientStreamTracer.java +++ b/core/src/main/java/io/grpc/internal/ForwardingClientStreamTracer.java @@ -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); diff --git a/core/src/main/java/io/grpc/internal/StatsTraceContext.java b/core/src/main/java/io/grpc/internal/StatsTraceContext.java index 007aefc0fb8..2827f6a9766 100644 --- a/core/src/main/java/io/grpc/internal/StatsTraceContext.java +++ b/core/src/main/java/io/grpc/internal/StatsTraceContext.java @@ -167,6 +167,19 @@ public void serverCallMethodResolved(MethodDescriptor method) { } } + /** + * See {@link ClientStreamTracer#cancelled}. For client-side only. + * + *

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. diff --git a/core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java b/core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java index 8f14b74035c..ddcf1bf44f4 100644 --- a/core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java +++ b/core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java @@ -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; @@ -155,6 +156,139 @@ 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 + verify(mockTracer, never()).cancelled(any(Status.class)); + } + + @Test + public void transportReportStatus_stopDeliveryFalse_doesNotNotifyTracerCancelled() { + 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() {}, + customStatsTraceCtx, transportTracer); + stream.start(mockListener); + + // Server-initiated CANCELLED (stopDelivery = false) + state.transportReportStatus(Status.CANCELLED, false, new Metadata()); + + verify(mockTracer, never()).cancelled(any(Status.class)); + verify(mockTracer).streamClosed(Status.CANCELLED); + } + + @Test + public void transportReportStatus_stopDeliveryTrue_notifiesTracerCancelled() { + 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() {}, + customStatsTraceCtx, transportTracer); + stream.start(mockListener); + + // Client/Transport-initiated cancellation (stopDelivery = true) + Status cancelStatus = Status.CANCELLED.withDescription("Client cancelled"); + state.transportReportStatus(cancelStatus, true, new Metadata()); + + verify(mockTracer).cancelled(cancelStatus); + verify(mockTracer).streamClosed(cancelStatus); + } + + @Test + public void transportReportStatus_stopDeliveryFalse_deadlineExceeded_noTracerCancelled() { + 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() {}, customStatsTraceCtx, transportTracer); + stream.start(mockListener); + + // Server-initiated DEADLINE_EXCEEDED (stopDelivery = false) + Status status = Status.DEADLINE_EXCEEDED.withDescription("Server deadline exceeded"); + state.transportReportStatus(status, false, new Metadata()); + + verify(mockTracer, never()).cancelled(any(Status.class)); + verify(mockTracer).streamClosed(status); + } + + @Test + public void transportReportStatus_stopDeliveryTrue_deadlineExceeded_notifiesTracerCancelled() { + 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() {}, customStatsTraceCtx, transportTracer); + stream.start(mockListener); + + // Client/Transport-initiated deadline exceeded (stopDelivery = true) + Status status = Status.DEADLINE_EXCEEDED.withDescription("Client deadline exceeded"); + state.transportReportStatus(status, true, new Metadata()); + + verify(mockTracer).cancelled(status); + verify(mockTracer).streamClosed(status); + } + + @Test + public void closeListener_deferredDeframerClose_stopDeliveryFalse_delaysCloseListener() { + ClientStreamTracer mockTracer = mock(ClientStreamTracer.class); + StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer}); + BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer); + AbstractClientStream stream = new BaseAbstractClientStream( + allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer); + stream.start(mockListener); + + // Send partial message into deframer + byte[] data = new byte[] {0, 0, 0, 0, 2, 1}; // 2-byte frame, only 1 byte delivered + state.deframe(ReadableBuffers.wrap(data)); + + Status statusFalse = Status.CANCELLED.withDescription("deferred stopDelivery false"); + state.transportReportStatus(statusFalse, false, new Metadata()); + + // Listener is not closed yet because deframer is mid-frame and waiting for complete frame + verify(mockTracer, never()).cancelled(any(Status.class)); + verify(mockTracer, never()).streamClosed(any(Status.class)); + + // Request message and provide remaining byte of frame to complete deframer processing + stream.request(1); + state.deframe(ReadableBuffers.wrap(new byte[] {2})); + verify(mockTracer, never()).cancelled(any(Status.class)); + verify(mockTracer).streamClosed(any(Status.class)); + } + @Test public void startFailsOnNullListener() { AbstractClientStream stream = diff --git a/core/src/test/java/io/grpc/internal/StatsTraceContextTest.java b/core/src/test/java/io/grpc/internal/StatsTraceContextTest.java new file mode 100644 index 00000000000..c00efd9e8de --- /dev/null +++ b/core/src/test/java/io/grpc/internal/StatsTraceContextTest.java @@ -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); + } +} diff --git a/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java b/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java index a92f10fd5c5..6c11229fa3d 100644 --- a/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java +++ b/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java @@ -846,6 +846,7 @@ public void cancel(Status reason) { if (!internalCancel(serverStatus, serverStatus)) { return; } + statsTraceCtx.clientCancelled(reason); serverStream.clientCancelled(reason); streamClosed(); } diff --git a/inprocess/src/test/java/io/grpc/inprocess/InProcessTransportTest.java b/inprocess/src/test/java/io/grpc/inprocess/InProcessTransportTest.java index d2220e05114..6955e749469 100644 --- a/inprocess/src/test/java/io/grpc/inprocess/InProcessTransportTest.java +++ b/inprocess/src/test/java/io/grpc/inprocess/InProcessTransportTest.java @@ -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; @@ -236,6 +237,144 @@ 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)); + } + + @Test + public void serverStream_closeWithCancelled_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; + + Status serverStatus = Status.CANCELLED.withDescription("Server cancelled over wire"); + serverStream.close(serverStatus, new Metadata()); + + org.mockito.Mockito.verify(mockTracer, org.mockito.Mockito.never()) + .cancelled(org.mockito.Mockito.any(Status.class)); + org.mockito.ArgumentCaptor statusCaptor = + org.mockito.ArgumentCaptor.forClass(Status.class); + org.mockito.Mockito.verify(mockTracer).streamClosed(statusCaptor.capture()); + assertEquals(Status.Code.CANCELLED, statusCaptor.getValue().getCode()); + assertEquals("Server cancelled over wire", statusCaptor.getValue().getDescription()); + } + + @Test + public void serverStream_closeWithDeadlineExceeded_noTracerCancelled() 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; + + Status serverStatus = Status.DEADLINE_EXCEEDED.withDescription("Server deadline exceeded"); + serverStream.close(serverStatus, new Metadata()); + + org.mockito.Mockito.verify(mockTracer, org.mockito.Mockito.never()) + .cancelled(org.mockito.Mockito.any(Status.class)); + org.mockito.ArgumentCaptor statusCaptor = + org.mockito.ArgumentCaptor.forClass(Status.class); + org.mockito.Mockito.verify(mockTracer).streamClosed(statusCaptor.capture()); + assertEquals(Status.Code.DEADLINE_EXCEEDED, statusCaptor.getValue().getCode()); + assertEquals("Server deadline exceeded", statusCaptor.getValue().getDescription()); + } + + @Test + public void clientStream_cancelWithDeadlineExceeded_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.DEADLINE_EXCEEDED.withDescription("Client deadline exceeded"); + clientStream.cancel(cancelStatus); + + org.mockito.Mockito.verify(mockTracer).cancelled(cancelStatus); + org.mockito.Mockito.verify(mockTracer).streamClosed(cancelStatus); + } + private void assertAssumedMessageSize( TestStreamTracer streamTracerSender, TestStreamTracer streamTracerReceiver) { if (isEnabledSupportTracingMessageSizes()) { diff --git a/util/src/main/java/io/grpc/util/ForwardingClientStreamTracer.java b/util/src/main/java/io/grpc/util/ForwardingClientStreamTracer.java index 9c9998571e5..1eda7c4bdd2 100644 --- a/util/src/main/java/io/grpc/util/ForwardingClientStreamTracer.java +++ b/util/src/main/java/io/grpc/util/ForwardingClientStreamTracer.java @@ -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); diff --git a/util/src/main/java/io/grpc/util/OutlierDetectionLoadBalancer.java b/util/src/main/java/io/grpc/util/OutlierDetectionLoadBalancer.java index dc61441bccd..b7fd50ae6cd 100644 --- a/util/src/main/java/io/grpc/util/OutlierDetectionLoadBalancer.java +++ b/util/src/main/java/io/grpc/util/OutlierDetectionLoadBalancer.java @@ -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()); + } } }; } diff --git a/util/src/test/java/io/grpc/util/OutlierDetectionLoadBalancerTest.java b/util/src/test/java/io/grpc/util/OutlierDetectionLoadBalancerTest.java index 39f5b5fb7d6..2e1aabfb5a9 100644 --- a/util/src/test/java/io/grpc/util/OutlierDetectionLoadBalancerTest.java +++ b/util/src/test/java/io/grpc/util/OutlierDetectionLoadBalancerTest.java @@ -451,6 +451,29 @@ public void delegatePickTracerFactoryPreserved() { verify(mockStreamTracer).inboundHeaders(); } + @Test + public void delegatePick_cancelledForwarded() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setSuccessRateEjection(new SuccessRateEjection.Builder().build()) + .setChildConfig(newChildConfig(fakeLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers.get(0))); + + final Subchannel readySubchannel = subchannels.values().iterator().next(); + deliverSubchannelState(readySubchannel, ConnectivityStateInfo.forNonError(READY)); + + verify(mockHelper, times(2)).updateBalancingState(stateCaptor.capture(), + pickerCaptor.capture()); + + SubchannelPicker picker = pickerCaptor.getAllValues().get(1); + PickResult pickResult = picker.pickSubchannel(mock(PickSubchannelArgs.class)); + + ClientStreamTracer clientStreamTracer = pickResult.getStreamTracerFactory() + .newClientStreamTracer(ClientStreamTracer.StreamInfo.newBuilder().build(), new Metadata()); + clientStreamTracer.cancelled(Status.CANCELLED); + verify(mockStreamTracer).cancelled(Status.CANCELLED); + } + /** * Assure the tracer works even when the underlying LB does not have a tracer to delegate to. */ @@ -522,7 +545,7 @@ public void successRateOneOutlier() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -531,6 +554,108 @@ public void successRateOneOutlier() { assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses()))); } + /** + * The success rate algorithm ignores CANCELLED status calls. + */ + @Test + public void successRateOneOutlier_cancelledIgnored() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setSuccessRateEjection( + new SuccessRateEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + // subchannel1 returns CANCELLED. + generateLoad(ImmutableMap.of(subchannel1, Status.CANCELLED), 7); + + // Move forward in time to a point where the detection timer has fired. + forwardTime(config); + + // CANCELLED status should be excluded from call counting, so no ejections occur. + assertEjectedSubchannels(ImmutableSet.of()); + } + + /** + * The success rate algorithm ignores DEADLINE_EXCEEDED status calls. + */ + @Test + public void successRateOneOutlier_deadlineExceededIgnored() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setSuccessRateEjection( + new SuccessRateEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + // subchannel1 returns DEADLINE_EXCEEDED. + generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + + // Move forward in time to a point where the detection timer has fired. + forwardTime(config); + + // DEADLINE_EXCEEDED status should be excluded from call counting, so no ejections occur. + assertEjectedSubchannels(ImmutableSet.of()); + } + + /** + * Server-initiated CANCELLED status over the wire (stopDelivery = false) counts as failure + * and results in ejection under success rate algorithm. + */ + @Test + public void successRateOneOutlier_serverInitiatedCancelledEjected() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setSuccessRateEjection( + new SuccessRateEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + // subchannel1 returns CANCELLED from server (no client cancellation tracer callback). + generateServerInitiatedLoad(ImmutableMap.of(subchannel1, Status.CANCELLED), 7); + + // Move forward in time to a point where the detection timer has fired. + forwardTime(config); + + // Server-initiated CANCELLED status is counted as a failure, so subchannel1 should be ejected. + assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses()))); + } + + /** + * Server-initiated DEADLINE_EXCEEDED status over the wire (stopDelivery = false) counts + * as failure and results in ejection under success rate algorithm. + */ + @Test + public void successRateOneOutlier_serverInitiatedDeadlineExceededEjected() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setSuccessRateEjection( + new SuccessRateEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + // subchannel1 returns DEADLINE_EXCEEDED from server (no client cancellation tracer callback). + generateServerInitiatedLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + + // Move forward in time to a point where the detection timer has fired. + forwardTime(config); + + // Server-initiated DEADLINE_EXCEEDED status is counted as a failure, so subchannel1 is ejected. + assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses()))); + } + /** * The success rate algorithm ejects the outlier, but then the config changes so that similar * behavior no longer gets ejected. @@ -547,7 +672,7 @@ public void successRateOneOutlier_configChange() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -567,7 +692,7 @@ public void successRateOneOutlier_configChange() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel2, Status.DEADLINE_EXCEEDED), 8); + generateLoad(ImmutableMap.of(subchannel2, Status.UNAVAILABLE), 8); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -592,7 +717,7 @@ public void successRateOneOutlier_unejected() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. fakeClock.forwardTime(config.intervalNanos + 1, TimeUnit.NANOSECONDS); @@ -627,7 +752,7 @@ public void successRateOneOutlier_notEnoughVolume() { // We produce an outlier, but don't give it enough calls to reach the minimum volume. generateLoad( - ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), + ImmutableMap.of(subchannel1, Status.UNAVAILABLE), ImmutableMap.of(subchannel1, 19), 7); // Move forward in time to a point where the detection timer has fired. @@ -654,7 +779,7 @@ public void successRateOneOutlier_notEnoughAddressesWithVolume() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); generateLoad( - ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), + ImmutableMap.of(subchannel1, Status.UNAVAILABLE), // subchannel2 has only 19 calls which results in success rate not triggering. ImmutableMap.of(subchannel2, 19), 7); @@ -683,7 +808,7 @@ public void successRateOneOutlier_enforcementPercentage() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -709,8 +834,8 @@ public void successRateTwoOutliers() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); generateLoad(ImmutableMap.of( - subchannel1, Status.DEADLINE_EXCEEDED, - subchannel2, Status.DEADLINE_EXCEEDED), 7); + subchannel1, Status.UNAVAILABLE, + subchannel2, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -738,9 +863,9 @@ public void successRateThreeOutliers_maxEjectionPercentage() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); generateLoad(ImmutableMap.of( - subchannel1, Status.DEADLINE_EXCEEDED, - subchannel2, Status.DEADLINE_EXCEEDED, - subchannel3, Status.DEADLINE_EXCEEDED), 7); + subchannel1, Status.UNAVAILABLE, + subchannel2, Status.UNAVAILABLE, + subchannel3, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -781,6 +906,52 @@ public void failurePercentageNoOutliers() { assertEjectedSubchannels(ImmutableSet.of()); } + /** + * Client-cancelled streams (e.g. non-winning hedged attempts) do not count as failures for + * failure percentage algorithm. + */ + @Test + public void failurePercentage_clientCancelled_notEjected() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setFailurePercentageEjection( + new FailurePercentageEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(READY)); + deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(READY)); + deliverSubchannelState(subchannel3, ConnectivityStateInfo.forNonError(READY)); + deliverSubchannelState(subchannel4, ConnectivityStateInfo.forNonError(READY)); + deliverSubchannelState(subchannel5, ConnectivityStateInfo.forNonError(READY)); + + verify(mockHelper, times(7)).updateBalancingState(stateCaptor.capture(), + pickerCaptor.capture()); + SubchannelPicker picker = pickerCaptor.getAllValues() + .get(pickerCaptor.getAllValues().size() - 1); + + for (int i = 0; i < 100; i++) { + PickResult pickResult = picker.pickSubchannel(mock(PickSubchannelArgs.class)); + ClientStreamTracer clientStreamTracer = pickResult.getStreamTracerFactory() + .newClientStreamTracer(null, null); + Subchannel subchannel = (Subchannel) pickResult.getSubchannel().getInternalSubchannel(); + if (subchannel == subchannel1) { + clientStreamTracer.cancelled(Status.CANCELLED); + clientStreamTracer.streamClosed(Status.CANCELLED); + } else { + clientStreamTracer.streamClosed(Status.OK); + } + } + + forwardTime(config); + + // subchannel1 was cancelled client-side and should not be ejected as an outlier. + assertEjectedSubchannels(ImmutableSet.of()); + } + /** * The success rate algorithm ejects the outlier. */ @@ -796,7 +967,7 @@ public void failurePercentageOneOutlier() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -805,6 +976,108 @@ public void failurePercentageOneOutlier() { assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses()))); } + /** + * The failure percentage algorithm ignores CANCELLED status calls. + */ + @Test + public void failurePercentageOneOutlier_cancelledIgnored() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setFailurePercentageEjection( + new FailurePercentageEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + // subchannel1 returns CANCELLED. + generateLoad(ImmutableMap.of(subchannel1, Status.CANCELLED), 7); + + // Move forward in time to a point where the detection timer has fired. + forwardTime(config); + + // CANCELLED status should be excluded from call counting, so no ejections occur. + assertEjectedSubchannels(ImmutableSet.of()); + } + + /** + * The failure percentage algorithm ignores DEADLINE_EXCEEDED status calls. + */ + @Test + public void failurePercentageOneOutlier_deadlineExceededIgnored() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setFailurePercentageEjection( + new FailurePercentageEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + // subchannel1 returns DEADLINE_EXCEEDED. + generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + + // Move forward in time to a point where the detection timer has fired. + forwardTime(config); + + // DEADLINE_EXCEEDED status should be excluded from call counting, so no ejections occur. + assertEjectedSubchannels(ImmutableSet.of()); + } + + /** + * Server-initiated CANCELLED status over the wire (stopDelivery = false) counts as failure + * and results in ejection under failure percentage algorithm. + */ + @Test + public void failurePercentageOneOutlier_serverInitiatedCancelledEjected() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setFailurePercentageEjection( + new FailurePercentageEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + // subchannel1 returns CANCELLED from server (no client cancellation tracer callback). + generateServerInitiatedLoad(ImmutableMap.of(subchannel1, Status.CANCELLED), 7); + + // Move forward in time to a point where the detection timer has fired. + forwardTime(config); + + // Server-initiated CANCELLED status is counted as a failure, so subchannel1 should be ejected. + assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses()))); + } + + /** + * Server-initiated DEADLINE_EXCEEDED status over the wire (stopDelivery = false) counts + * as failure and results in ejection under failure percentage algorithm. + */ + @Test + public void failurePercentageOneOutlier_serverInitiatedDeadlineExceededEjected() { + OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder() + .setMaxEjectionPercent(50) + .setFailurePercentageEjection( + new FailurePercentageEjection.Builder() + .setMinimumHosts(3) + .setRequestVolume(10).build()) + .setChildConfig(newChildConfig(roundRobinLbProvider, null)).build(); + + loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); + + // subchannel1 returns DEADLINE_EXCEEDED from server (no client cancellation tracer callback). + generateServerInitiatedLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + + // Move forward in time to a point where the detection timer has fired. + forwardTime(config); + + // Server-initiated DEADLINE_EXCEEDED status is counted as a failure, so subchannel1 is ejected. + assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses()))); + } + /** * The failure percentage algorithm ignores addresses without enough volume.. */ @@ -820,7 +1093,7 @@ public void failurePercentageOneOutlier_notEnoughVolume() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -846,7 +1119,7 @@ public void failurePercentageOneOutlier_notEnoughAddressesWithVolume() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); generateLoad( - ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), + ImmutableMap.of(subchannel1, Status.UNAVAILABLE), // subchannel2 has only 19 calls which results in failure percentage not triggering. ImmutableMap.of(subchannel2, 19), 7); @@ -875,7 +1148,7 @@ public void failurePercentageOneOutlier_enforcementPercentage() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -909,9 +1182,9 @@ public void successRateAndFailurePercentageThreeOutliers() { // configured with a 0 tolerance threshold. generateLoad( ImmutableMap.of( - subchannel1, Status.DEADLINE_EXCEEDED, - subchannel2, Status.DEADLINE_EXCEEDED, - subchannel3, Status.DEADLINE_EXCEEDED), + subchannel1, Status.UNAVAILABLE, + subchannel2, Status.UNAVAILABLE, + subchannel3, Status.UNAVAILABLE), ImmutableMap.of(subchannel3, 1), 7); // Move forward in time to a point where the detection timer has fired. @@ -941,7 +1214,7 @@ public void subchannelUpdateAddress_singleReplaced() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -1003,8 +1276,8 @@ public void multipleAddressesEndpoint() { assertThat(loadBalancer.endpointTrackerMap.size()).isEqualTo(3); assertThat(loadBalancer.addressMap.size()).isEqualTo(5); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED, - subchannel2, Status.DEADLINE_EXCEEDED), 13); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE, + subchannel2, Status.UNAVAILABLE), 13); forwardTime(config); // eject the first endpoint: (address0, address1) @@ -1072,7 +1345,7 @@ public void subchannelUpdateAddress_multipleReplacedWithSingle() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 6); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 6); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -1162,7 +1435,7 @@ public void successRateAndFailurePercentage_successRateOutlier() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -1191,7 +1464,7 @@ public void successRateAndFailurePercentage_successRateOutlier_() { // with heal loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 6); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 6); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -1235,7 +1508,7 @@ public void successRateAndFailurePercentage_errorPercentageOutlier() { loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 7); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -1264,7 +1537,7 @@ public void successRateAndFailurePercentage_errorPercentageOutlier_() { // with loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers)); - generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 6); + generateLoad(ImmutableMap.of(subchannel1, Status.UNAVAILABLE), 6); // Move forward in time to a point where the detection timer has fired. forwardTime(config); @@ -1363,12 +1636,46 @@ private void generateLoad(Map statusMap, int calls = callCountMap.containsKey(subchannel) ? callCountMap.get(subchannel) : 0; if (calls < maxCalls) { callCountMap.put(subchannel, ++calls); - clientStreamTracer.streamClosed( - statusMap.containsKey(subchannel) ? statusMap.get(subchannel) : Status.OK); + Status status = statusMap.containsKey(subchannel) ? statusMap.get(subchannel) : Status.OK; + if (status.getCode() == Status.Code.CANCELLED + || status.getCode() == Status.Code.DEADLINE_EXCEEDED) { + clientStreamTracer.cancelled(status); + } + clientStreamTracer.streamClosed(status); } } } + // Generates 100 calls, simulating server-initiated status responses over the wire. + private void generateServerInitiatedLoad( + Map statusMap, int expectedStateChanges) { + deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(READY)); + deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(READY)); + deliverSubchannelState(subchannel3, ConnectivityStateInfo.forNonError(READY)); + deliverSubchannelState(subchannel4, ConnectivityStateInfo.forNonError(READY)); + deliverSubchannelState(subchannel5, ConnectivityStateInfo.forNonError(READY)); + + verify(mockHelper, times(expectedStateChanges)).updateBalancingState(stateCaptor.capture(), + pickerCaptor.capture()); + SubchannelPicker picker = pickerCaptor.getAllValues() + .get(pickerCaptor.getAllValues().size() - 1); + + HashMap callCountMap = new HashMap<>(); + for (int i = 0; i < 100; i++) { + PickResult pickResult = picker + .pickSubchannel(mock(PickSubchannelArgs.class)); + ClientStreamTracer clientStreamTracer = pickResult.getStreamTracerFactory() + .newClientStreamTracer(null, null); + + Subchannel subchannel = (Subchannel) pickResult.getSubchannel().getInternalSubchannel(); + + int calls = callCountMap.containsKey(subchannel) ? callCountMap.get(subchannel) : 0; + callCountMap.put(subchannel, ++calls); + Status status = statusMap.containsKey(subchannel) ? statusMap.get(subchannel) : Status.OK; + clientStreamTracer.streamClosed(status); + } + } + // Forwards time past the moment when the timer will fire. private void forwardTime(OutlierDetectionLoadBalancerConfig config) { fakeClock.forwardTime(config.intervalNanos + 1, TimeUnit.NANOSECONDS);