Skip to content

Commit be50f7b

Browse files
AgraVatorAgraVator
committed
test: add comprehensive stopDelivery and server vs client cancellation unit tests
1 parent 5f3af3d commit be50f7b

3 files changed

Lines changed: 312 additions & 0 deletions

File tree

‎core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -228,6 +228,96 @@ public void transportReportStatus_stopDeliveryTrue_notifiesTracerCancelled() {
228228
verify(mockTracer).streamClosed(cancelStatus);
229229
}
230230

231+
@Test
232+
public void transportReportStatus_stopDeliveryFalse_deadlineExceeded_noTracerCancelled() {
233+
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
234+
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
235+
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
236+
AbstractClientStream stream = new BaseAbstractClientStream(
237+
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
238+
stream.start(mockListener);
239+
240+
// Server-initiated DEADLINE_EXCEEDED (stopDelivery = false)
241+
Status status = Status.DEADLINE_EXCEEDED.withDescription("Server deadline exceeded");
242+
state.transportReportStatus(status, false, new Metadata());
243+
244+
verify(mockTracer, never()).cancelled(any(Status.class));
245+
verify(mockTracer).streamClosed(status);
246+
}
247+
248+
@Test
249+
public void transportReportStatus_stopDeliveryTrue_deadlineExceeded_notifiesTracerCancelled() {
250+
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
251+
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
252+
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
253+
AbstractClientStream stream = new BaseAbstractClientStream(
254+
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
255+
stream.start(mockListener);
256+
257+
// Client/Transport-initiated deadline exceeded (stopDelivery = true)
258+
Status status = Status.DEADLINE_EXCEEDED.withDescription("Client deadline exceeded");
259+
state.transportReportStatus(status, true, new Metadata());
260+
261+
verify(mockTracer).cancelled(status);
262+
verify(mockTracer).streamClosed(status);
263+
}
264+
265+
@Test
266+
public void closeListener_directAssertions_stopDeliveryTrueAndFalse() {
267+
ClientStreamTracer mockTracer1 = mock(ClientStreamTracer.class);
268+
StatsTraceContext statsTraceCtx1 = new StatsTraceContext(new StreamTracer[] {mockTracer1});
269+
BaseTransportState state1 = new BaseTransportState(statsTraceCtx1, transportTracer);
270+
AbstractClientStream stream1 = new BaseAbstractClientStream(
271+
allocator, state1, new BaseSink() {}, statsTraceCtx1, transportTracer);
272+
stream1.start(mockListener);
273+
274+
// stopDelivery = true: clientCancelled is called before streamClosed
275+
Status statusTrue = Status.CANCELLED.withDescription("stopDelivery true");
276+
state1.transportReportStatus(statusTrue, true, new Metadata());
277+
verify(mockTracer1).cancelled(statusTrue);
278+
verify(mockTracer1).streamClosed(statusTrue);
279+
280+
ClientStreamTracer mockTracer2 = mock(ClientStreamTracer.class);
281+
StatsTraceContext statsTraceCtx2 = new StatsTraceContext(new StreamTracer[] {mockTracer2});
282+
BaseTransportState state2 = new BaseTransportState(statsTraceCtx2, transportTracer);
283+
AbstractClientStream stream2 = new BaseAbstractClientStream(
284+
allocator, state2, new BaseSink() {}, statsTraceCtx2, transportTracer);
285+
stream2.start(mockListener);
286+
287+
// stopDelivery = false: clientCancelled is NOT called, only streamClosed
288+
Status statusFalse = Status.CANCELLED.withDescription("stopDelivery false");
289+
state2.transportReportStatus(statusFalse, false, new Metadata());
290+
verify(mockTracer2, never()).cancelled(any(Status.class));
291+
verify(mockTracer2).streamClosed(statusFalse);
292+
}
293+
294+
@Test
295+
public void closeListener_deferredDeframerClose_stopDeliveryFalse_delaysCloseListener() {
296+
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
297+
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
298+
BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
299+
AbstractClientStream stream = new BaseAbstractClientStream(
300+
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
301+
stream.start(mockListener);
302+
303+
// Send partial message into deframer
304+
byte[] data = new byte[] {0, 0, 0, 0, 2, 1}; // 2-byte frame, only 1 byte delivered
305+
state.deframe(ReadableBuffers.wrap(data));
306+
307+
Status statusFalse = Status.CANCELLED.withDescription("deferred stopDelivery false");
308+
state.transportReportStatus(statusFalse, false, new Metadata());
309+
310+
// Listener is not closed yet because deframer is mid-frame and waiting for complete frame
311+
verify(mockTracer, never()).cancelled(any(Status.class));
312+
verify(mockTracer, never()).streamClosed(any(Status.class));
313+
314+
// Request message and provide remaining byte of frame to complete deframer processing
315+
stream.request(1);
316+
state.deframe(ReadableBuffers.wrap(new byte[] {2}));
317+
verify(mockTracer, never()).cancelled(any(Status.class));
318+
verify(mockTracer).streamClosed(any(Status.class));
319+
}
320+
231321
@Test
232322
public void startFailsOnNullListener() {
233323
AbstractClientStream stream =

‎inprocess/src/test/java/io/grpc/inprocess/InProcessTransportTest.java‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -287,6 +287,94 @@ methodDescriptor, new Metadata(), CallOptions.DEFAULT,
287287
.cancelled(org.mockito.Mockito.any(Status.class));
288288
}
289289

290+
@Test
291+
public void serverStream_closeWithCancelled_doesNotNotifyTracerCancelled() throws Exception {
292+
server = newServer(Arrays.asList(serverStreamTracerFactory));
293+
server.start(serverListener);
294+
client = newClientTransport(server);
295+
startTransport(client, mockClientTransportListener);
296+
MockServerTransportListener serverTransportListener =
297+
serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
298+
serverTransport = serverTransportListener.transport;
299+
300+
ClientStreamTracer mockTracer = org.mockito.Mockito.mock(ClientStreamTracer.class);
301+
ClientStream clientStream = client.newStream(
302+
methodDescriptor, new Metadata(), CallOptions.DEFAULT,
303+
new ClientStreamTracer[] {mockTracer});
304+
ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
305+
clientStream.start(clientStreamListener);
306+
StreamCreation serverStreamCreation =
307+
serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
308+
ServerStream serverStream = serverStreamCreation.stream;
309+
310+
Status serverStatus = Status.CANCELLED.withDescription("Server cancelled over wire");
311+
serverStream.close(serverStatus, new Metadata());
312+
313+
org.mockito.Mockito.verify(mockTracer, org.mockito.Mockito.never())
314+
.cancelled(org.mockito.Mockito.any(Status.class));
315+
org.mockito.ArgumentCaptor<Status> statusCaptor =
316+
org.mockito.ArgumentCaptor.forClass(Status.class);
317+
org.mockito.Mockito.verify(mockTracer).streamClosed(statusCaptor.capture());
318+
assertEquals(Status.Code.CANCELLED, statusCaptor.getValue().getCode());
319+
assertEquals("Server cancelled over wire", statusCaptor.getValue().getDescription());
320+
}
321+
322+
@Test
323+
public void serverStream_closeWithDeadlineExceeded_noTracerCancelled() throws Exception {
324+
server = newServer(Arrays.asList(serverStreamTracerFactory));
325+
server.start(serverListener);
326+
client = newClientTransport(server);
327+
startTransport(client, mockClientTransportListener);
328+
MockServerTransportListener serverTransportListener =
329+
serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
330+
serverTransport = serverTransportListener.transport;
331+
332+
ClientStreamTracer mockTracer = org.mockito.Mockito.mock(ClientStreamTracer.class);
333+
ClientStream clientStream = client.newStream(
334+
methodDescriptor, new Metadata(), CallOptions.DEFAULT,
335+
new ClientStreamTracer[] {mockTracer});
336+
ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
337+
clientStream.start(clientStreamListener);
338+
StreamCreation serverStreamCreation =
339+
serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
340+
ServerStream serverStream = serverStreamCreation.stream;
341+
342+
Status serverStatus = Status.DEADLINE_EXCEEDED.withDescription("Server deadline exceeded");
343+
serverStream.close(serverStatus, new Metadata());
344+
345+
org.mockito.Mockito.verify(mockTracer, org.mockito.Mockito.never())
346+
.cancelled(org.mockito.Mockito.any(Status.class));
347+
org.mockito.ArgumentCaptor<Status> statusCaptor =
348+
org.mockito.ArgumentCaptor.forClass(Status.class);
349+
org.mockito.Mockito.verify(mockTracer).streamClosed(statusCaptor.capture());
350+
assertEquals(Status.Code.DEADLINE_EXCEEDED, statusCaptor.getValue().getCode());
351+
assertEquals("Server deadline exceeded", statusCaptor.getValue().getDescription());
352+
}
353+
354+
@Test
355+
public void clientStream_cancelWithDeadlineExceeded_notifiesTracerCancelled() throws Exception {
356+
server = newServer(Arrays.asList(serverStreamTracerFactory));
357+
server.start(serverListener);
358+
client = newClientTransport(server);
359+
startTransport(client, mockClientTransportListener);
360+
MockServerTransportListener serverTransportListener =
361+
serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
362+
serverTransport = serverTransportListener.transport;
363+
364+
ClientStreamTracer mockTracer = org.mockito.Mockito.mock(ClientStreamTracer.class);
365+
ClientStream clientStream = client.newStream(
366+
methodDescriptor, new Metadata(), CallOptions.DEFAULT,
367+
new ClientStreamTracer[] {mockTracer});
368+
ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
369+
clientStream.start(clientStreamListener);
370+
371+
Status cancelStatus = Status.DEADLINE_EXCEEDED.withDescription("Client deadline exceeded");
372+
clientStream.cancel(cancelStatus);
373+
374+
org.mockito.Mockito.verify(mockTracer).cancelled(cancelStatus);
375+
org.mockito.Mockito.verify(mockTracer).streamClosed(cancelStatus);
376+
}
377+
290378
private void assertAssumedMessageSize(
291379
TestStreamTracer streamTracerSender, TestStreamTracer streamTracerReceiver) {
292380
if (isEnabledSupportTracingMessageSizes()) {

‎util/src/test/java/io/grpc/util/OutlierDetectionLoadBalancerTest.java‎

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -604,6 +604,58 @@ public void successRateOneOutlier_deadlineExceededIgnored() {
604604
assertEjectedSubchannels(ImmutableSet.of());
605605
}
606606

607+
/**
608+
* Server-initiated CANCELLED status over the wire (stopDelivery = false) counts as failure
609+
* and results in ejection under success rate algorithm.
610+
*/
611+
@Test
612+
public void successRateOneOutlier_serverInitiatedCancelledEjected() {
613+
OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder()
614+
.setMaxEjectionPercent(50)
615+
.setSuccessRateEjection(
616+
new SuccessRateEjection.Builder()
617+
.setMinimumHosts(3)
618+
.setRequestVolume(10).build())
619+
.setChildConfig(newChildConfig(roundRobinLbProvider, null)).build();
620+
621+
loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers));
622+
623+
// subchannel1 returns CANCELLED from server (no client cancellation tracer callback).
624+
generateServerInitiatedLoad(ImmutableMap.of(subchannel1, Status.CANCELLED), 7);
625+
626+
// Move forward in time to a point where the detection timer has fired.
627+
forwardTime(config);
628+
629+
// Server-initiated CANCELLED status is counted as a failure, so subchannel1 should be ejected.
630+
assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses())));
631+
}
632+
633+
/**
634+
* Server-initiated DEADLINE_EXCEEDED status over the wire (stopDelivery = false) counts
635+
* as failure and results in ejection under success rate algorithm.
636+
*/
637+
@Test
638+
public void successRateOneOutlier_serverInitiatedDeadlineExceededEjected() {
639+
OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder()
640+
.setMaxEjectionPercent(50)
641+
.setSuccessRateEjection(
642+
new SuccessRateEjection.Builder()
643+
.setMinimumHosts(3)
644+
.setRequestVolume(10).build())
645+
.setChildConfig(newChildConfig(roundRobinLbProvider, null)).build();
646+
647+
loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers));
648+
649+
// subchannel1 returns DEADLINE_EXCEEDED from server (no client cancellation tracer callback).
650+
generateServerInitiatedLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);
651+
652+
// Move forward in time to a point where the detection timer has fired.
653+
forwardTime(config);
654+
655+
// Server-initiated DEADLINE_EXCEEDED status is counted as a failure, so subchannel1 is ejected.
656+
assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses())));
657+
}
658+
607659
/**
608660
* The success rate algorithm ejects the outlier, but then the config changes so that similar
609661
* behavior no longer gets ejected.
@@ -974,6 +1026,58 @@ public void failurePercentageOneOutlier_deadlineExceededIgnored() {
9741026
assertEjectedSubchannels(ImmutableSet.of());
9751027
}
9761028

1029+
/**
1030+
* Server-initiated CANCELLED status over the wire (stopDelivery = false) counts as failure
1031+
* and results in ejection under failure percentage algorithm.
1032+
*/
1033+
@Test
1034+
public void failurePercentageOneOutlier_serverInitiatedCancelledEjected() {
1035+
OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder()
1036+
.setMaxEjectionPercent(50)
1037+
.setFailurePercentageEjection(
1038+
new FailurePercentageEjection.Builder()
1039+
.setMinimumHosts(3)
1040+
.setRequestVolume(10).build())
1041+
.setChildConfig(newChildConfig(roundRobinLbProvider, null)).build();
1042+
1043+
loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers));
1044+
1045+
// subchannel1 returns CANCELLED from server (no client cancellation tracer callback).
1046+
generateServerInitiatedLoad(ImmutableMap.of(subchannel1, Status.CANCELLED), 7);
1047+
1048+
// Move forward in time to a point where the detection timer has fired.
1049+
forwardTime(config);
1050+
1051+
// Server-initiated CANCELLED status is counted as a failure, so subchannel1 should be ejected.
1052+
assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses())));
1053+
}
1054+
1055+
/**
1056+
* Server-initiated DEADLINE_EXCEEDED status over the wire (stopDelivery = false) counts
1057+
* as failure and results in ejection under failure percentage algorithm.
1058+
*/
1059+
@Test
1060+
public void failurePercentageOneOutlier_serverInitiatedDeadlineExceededEjected() {
1061+
OutlierDetectionLoadBalancerConfig config = new OutlierDetectionLoadBalancerConfig.Builder()
1062+
.setMaxEjectionPercent(50)
1063+
.setFailurePercentageEjection(
1064+
new FailurePercentageEjection.Builder()
1065+
.setMinimumHosts(3)
1066+
.setRequestVolume(10).build())
1067+
.setChildConfig(newChildConfig(roundRobinLbProvider, null)).build();
1068+
1069+
loadBalancer.acceptResolvedAddresses(buildResolvedAddress(config, servers));
1070+
1071+
// subchannel1 returns DEADLINE_EXCEEDED from server (no client cancellation tracer callback).
1072+
generateServerInitiatedLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);
1073+
1074+
// Move forward in time to a point where the detection timer has fired.
1075+
forwardTime(config);
1076+
1077+
// Server-initiated DEADLINE_EXCEEDED status is counted as a failure, so subchannel1 is ejected.
1078+
assertEjectedSubchannels(ImmutableSet.of(ImmutableSet.copyOf(servers.get(0).getAddresses())));
1079+
}
1080+
9771081
/**
9781082
* The failure percentage algorithm ignores addresses without enough volume..
9791083
*/
@@ -1542,6 +1646,36 @@ private void generateLoad(Map<Subchannel, Status> statusMap,
15421646
}
15431647
}
15441648

1649+
// Generates 100 calls, simulating server-initiated status responses over the wire.
1650+
private void generateServerInitiatedLoad(
1651+
Map<Subchannel, Status> statusMap, int expectedStateChanges) {
1652+
deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(READY));
1653+
deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(READY));
1654+
deliverSubchannelState(subchannel3, ConnectivityStateInfo.forNonError(READY));
1655+
deliverSubchannelState(subchannel4, ConnectivityStateInfo.forNonError(READY));
1656+
deliverSubchannelState(subchannel5, ConnectivityStateInfo.forNonError(READY));
1657+
1658+
verify(mockHelper, times(expectedStateChanges)).updateBalancingState(stateCaptor.capture(),
1659+
pickerCaptor.capture());
1660+
SubchannelPicker picker = pickerCaptor.getAllValues()
1661+
.get(pickerCaptor.getAllValues().size() - 1);
1662+
1663+
HashMap<Subchannel, Integer> callCountMap = new HashMap<>();
1664+
for (int i = 0; i < 100; i++) {
1665+
PickResult pickResult = picker
1666+
.pickSubchannel(mock(PickSubchannelArgs.class));
1667+
ClientStreamTracer clientStreamTracer = pickResult.getStreamTracerFactory()
1668+
.newClientStreamTracer(null, null);
1669+
1670+
Subchannel subchannel = (Subchannel) pickResult.getSubchannel().getInternalSubchannel();
1671+
1672+
int calls = callCountMap.containsKey(subchannel) ? callCountMap.get(subchannel) : 0;
1673+
callCountMap.put(subchannel, ++calls);
1674+
Status status = statusMap.containsKey(subchannel) ? statusMap.get(subchannel) : Status.OK;
1675+
clientStreamTracer.streamClosed(status);
1676+
}
1677+
}
1678+
15451679
// Forwards time past the moment when the timer will fire.
15461680
private void forwardTime(OutlierDetectionLoadBalancerConfig config) {
15471681
fakeClock.forwardTime(config.intervalNanos + 1, TimeUnit.NANOSECONDS);

0 commit comments

Comments
 (0)