Skip to content

Commit fdcd2cf

Browse files
ran-suran-su
committed
xds: make ControlPlaneClient.shutdown() synchronous on syncContext
When a higher-priority xDS server responds while a lower-priority fallback ControlPlaneClient already has a DiscoveryResponse runnable queued in the shared SynchronizationContext, calling shutdownLowerPriorityCpcs() removes the fallback entry from serverLrsClientMap and invokes cpc.shutdown(). Because ControlPlaneClient.shutdown() previously scheduled its cleanup via syncContext.execute(...), the cleanup was placed at the tail of the queue behind the already-queued response runnable. When that queued response runnable executed, adsStream had not been closed yet, so handleResourceUpdate() looked up serverLrsClientMap.get(serverInfo) (now null) and threw a NullPointerException when calling controlPlaneClient.ackResponse(...). Fix this by: 1. Executing ControlPlaneClient.shutdown() directly on syncContext (enforcing syncContext.throwIfNotInThisSynchronizationContext()), which all callers already satisfy, so adsStream is closed immediately before any subsequently queued runnables run. 2. Adding a defensive null check for controlPlaneClient in XdsClientImpl.handleResourceUpdate().
1 parent 18f412e commit fdcd2cf

4 files changed

Lines changed: 162 additions & 14 deletions

File tree

‎xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -114,21 +114,21 @@ final class ControlPlaneClient {
114114
logger.log(XdsLogLevel.INFO, "Created");
115115
}
116116

117+
// Must be synchronized.
117118
void shutdown() {
118-
syncContext.execute(new Runnable() {
119-
@Override
120-
public void run() {
121-
shutdown = true;
122-
logger.log(XdsLogLevel.INFO, "Shutting down");
123-
if (adsStream != null) {
124-
adsStream.close(Status.CANCELLED.withDescription("shutdown").asException());
125-
}
126-
if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {
127-
rpcRetryTimer.cancel();
128-
}
129-
xdsTransport.shutdown();
130-
}
131-
});
119+
syncContext.throwIfNotInThisSynchronizationContext();
120+
if (shutdown) {
121+
return;
122+
}
123+
shutdown = true;
124+
logger.log(XdsLogLevel.INFO, "Shutting down");
125+
if (adsStream != null) {
126+
adsStream.close(Status.CANCELLED.withDescription("shutdown").asException());
127+
}
128+
if (rpcRetryTimer != null && rpcRetryTimer.isPending()) {
129+
rpcRetryTimer.cancel();
130+
}
131+
xdsTransport.shutdown();
132132
}
133133

134134
@Override

‎xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -574,6 +574,11 @@ private <T extends ResourceUpdate> void handleResourceUpdate(
574574
XdsResourceType.Args args, List<Any> resources, XdsResourceType<T> xdsResourceType,
575575
boolean isFirstResponse, ProcessingTracker processingTracker) {
576576
ControlPlaneClient controlPlaneClient = serverCpClientMap.get(args.serverInfo);
577+
if (controlPlaneClient == null) {
578+
logger.log(XdsLogLevel.DEBUG,
579+
"Couldn't find CPC for {0}, so skipping resource update", args.serverInfo);
580+
return;
581+
}
577582

578583
if (isFirstResponse) {
579584
shutdownLowerPriorityCpcs(controlPlaneClient);

‎xds/src/test/java/io/grpc/xds/XdsClientFallbackTest.java‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,17 +22,23 @@
2222
import static org.mockito.AdditionalAnswers.delegatesTo;
2323
import static org.mockito.ArgumentMatchers.any;
2424
import static org.mockito.ArgumentMatchers.argThat;
25+
import static org.mockito.Mockito.doAnswer;
2526
import static org.mockito.Mockito.inOrder;
2627
import static org.mockito.Mockito.mock;
2728
import static org.mockito.Mockito.never;
2829
import static org.mockito.Mockito.timeout;
2930
import static org.mockito.Mockito.times;
3031
import static org.mockito.Mockito.verify;
32+
import static org.mockito.Mockito.when;
3133

3234
import com.google.common.collect.ImmutableList;
3335
import com.google.common.collect.ImmutableMap;
36+
import com.google.protobuf.Any;
37+
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
38+
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
3439
import io.grpc.ChannelCredentials;
3540
import io.grpc.Grpc;
41+
import io.grpc.MethodDescriptor;
3642
import io.grpc.MetricRecorder;
3743
import io.grpc.Status;
3844
import io.grpc.StatusOr;
@@ -591,6 +597,111 @@ public void used_then_mainServerRestart_fallbackServerUp() {
591597
assertThat(getLrsServerInfo("localhost:" + fallbackServer.getServer().getPort())).isNull();
592598
}
593599

600+
@Test
601+
public void concurrentResponsesOnMainAndFallback_shutsDownFallbackWithoutNpe() {
602+
@SuppressWarnings("unchecked")
603+
XdsTransportFactory.StreamingCall<DiscoveryRequest, DiscoveryResponse> mainCall1 =
604+
mock(XdsTransportFactory.StreamingCall.class);
605+
@SuppressWarnings("unchecked")
606+
XdsTransportFactory.StreamingCall<DiscoveryRequest, DiscoveryResponse> mainCall2 =
607+
mock(XdsTransportFactory.StreamingCall.class);
608+
@SuppressWarnings("unchecked")
609+
XdsTransportFactory.StreamingCall<DiscoveryRequest, DiscoveryResponse> fallbackCall =
610+
mock(XdsTransportFactory.StreamingCall.class);
611+
when(mainCall1.isReady()).thenReturn(true);
612+
when(mainCall2.isReady()).thenReturn(true);
613+
when(fallbackCall.isReady()).thenReturn(true);
614+
615+
XdsTransportFactory.XdsTransport mainTransport = mock(XdsTransportFactory.XdsTransport.class);
616+
XdsTransportFactory.XdsTransport fallbackTransport =
617+
mock(XdsTransportFactory.XdsTransport.class);
618+
when(mainTransport.<DiscoveryRequest, DiscoveryResponse>createStreamingCall(
619+
any(), any(MethodDescriptor.Marshaller.class), any(MethodDescriptor.Marshaller.class)))
620+
.thenReturn(mainCall1, mainCall2);
621+
when(fallbackTransport.<DiscoveryRequest, DiscoveryResponse>createStreamingCall(
622+
any(), any(MethodDescriptor.Marshaller.class), any(MethodDescriptor.Marshaller.class)))
623+
.thenReturn(fallbackCall);
624+
625+
XdsTransportFactory customTransportFactory =
626+
serverInfo ->
627+
serverInfo.target().equals("main-uri") ? mainTransport : fallbackTransport;
628+
629+
XdsClientImpl client =
630+
CommonBootstrapperTestUtils.createXdsClient(
631+
Arrays.asList("main-uri", "fallback-uri"),
632+
customTransportFactory,
633+
fakeClock,
634+
new ExponentialBackoffPolicy.Provider(),
635+
MessagePrinter.INSTANCE,
636+
xdsClientMetricReporter);
637+
638+
client.watchXdsResource(XdsListenerResource.getInstance(), MAIN_SERVER, ldsWatcher);
639+
640+
@SuppressWarnings("unchecked")
641+
ArgumentCaptor<XdsTransportFactory.EventHandler<DiscoveryResponse>> mainHandler1Captor =
642+
ArgumentCaptor.forClass(XdsTransportFactory.EventHandler.class);
643+
verify(mainCall1).start(mainHandler1Captor.capture());
644+
645+
// Fail main's initial ADS stream before any response, activating fallback.
646+
mainHandler1Captor.getValue().onStatusReceived(Status.UNAVAILABLE);
647+
648+
@SuppressWarnings("unchecked")
649+
ArgumentCaptor<XdsTransportFactory.EventHandler<DiscoveryResponse>> fallbackHandlerCaptor =
650+
ArgumentCaptor.forClass(XdsTransportFactory.EventHandler.class);
651+
verify(fallbackCall).start(fallbackHandlerCaptor.capture());
652+
fallbackHandlerCaptor.getValue().onReady();
653+
654+
// Advance clock so main's retry timer fires and opens a second ADS stream while fallback's
655+
// stream is also open.
656+
fakeClock.forwardTime(2, TimeUnit.SECONDS);
657+
658+
@SuppressWarnings("unchecked")
659+
ArgumentCaptor<XdsTransportFactory.EventHandler<DiscoveryResponse>> mainHandler2Captor =
660+
ArgumentCaptor.forClass(XdsTransportFactory.EventHandler.class);
661+
verify(mainCall2).start(mainHandler2Captor.capture());
662+
663+
DiscoveryResponse mainResponse =
664+
DiscoveryResponse.newBuilder()
665+
.setTypeUrl(ADS_TYPE_URL_LDS)
666+
.setVersionInfo("1")
667+
.setNonce("nonce-main")
668+
.addResources(Any.pack(ControlPlaneRule.buildClientListener(MAIN_SERVER, RDS_NAME)))
669+
.build();
670+
DiscoveryResponse fallbackResponse =
671+
DiscoveryResponse.newBuilder()
672+
.setTypeUrl(ADS_TYPE_URL_LDS)
673+
.setVersionInfo("1")
674+
.setNonce("nonce-fallback")
675+
.addResources(
676+
Any.pack(ControlPlaneRule.buildClientListener(MAIN_SERVER, FALLBACK_RDS_NAME)))
677+
.build();
678+
679+
// When mainCall2 sends its initial DiscoveryRequest (inside XdsClientImpl's syncContext),
680+
// enqueue onRecvMessage for both main and fallback before syncContext finishes draining.
681+
// Main's first response runs shutdownLowerPriorityCpcs(mainCpc), and fallback's already-queued
682+
// onRecvMessage runs immediately after in the same drain loop.
683+
doAnswer(
684+
invocation -> {
685+
mainHandler2Captor.getValue().onRecvMessage(mainResponse);
686+
fallbackHandlerCaptor.getValue().onRecvMessage(fallbackResponse);
687+
return null;
688+
})
689+
.doNothing()
690+
.when(mainCall2)
691+
.sendMessage(any());
692+
693+
mainHandler2Captor.getValue().onReady();
694+
695+
verify(ldsWatcher, timeout(5000))
696+
.onResourceChanged(
697+
StatusOr.fromValue(LdsUpdate.forApiListener(MAIN_HTTP_CONNECTION_MANAGER)));
698+
verify(fallbackCall).sendError(any());
699+
verify(fallbackTransport).shutdown();
700+
verify(fallbackCall, times(1)).sendMessage(any());
701+
702+
client.shutdown();
703+
}
704+
594705
/**
595706
* gRFC A95: for an authority with fallback_on_reachability_only, falling back must happen when
596707
* the primary server becomes unreachable, even though every subscribed resource is cached.

‎xds/src/test/java/io/grpc/xds/client/ControlPlaneClientTest.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -273,6 +273,38 @@ public void cancelBeforeAck_sendsEmptyUnsubscribe() {
273273
assertThat(unsub.getResourceNamesList()).isEmpty();
274274
}
275275

276+
@Test
277+
public void shutdown_closesStreamImmediatelyAndSkipsQueuedRecvMessage() {
278+
Map<String, XdsResourceType<?>> subscribedTypes = ImmutableMap.of(CDS_TYPE_URL, cdsType);
279+
when(resourceStore.getSubscribedResourceTypesWithTypeUrl()).thenReturn(subscribedTypes);
280+
when(resourceStore.getSubscribedResources(serverInfo, cdsType))
281+
.thenReturn(ImmutableList.of("foo-cluster"));
282+
283+
syncContext.execute(cpc::sendDiscoveryRequests);
284+
verify(streamingCall).start(handlerCaptor.capture());
285+
handlerCaptor.getValue().onReady();
286+
287+
DiscoveryResponse response =
288+
DiscoveryResponse.newBuilder()
289+
.setTypeUrl(CDS_TYPE_URL)
290+
.setVersionInfo("1")
291+
.setNonce("nonce-1")
292+
.build();
293+
294+
// Simulate a response arriving while syncContext is draining a task that shuts down this CPC
295+
// (e.g. XdsClientImpl.shutdownLowerPriorityCpcs when a higher-priority server responds first).
296+
syncContext.execute(
297+
() -> {
298+
handlerCaptor.getValue().onRecvMessage(response);
299+
cpc.shutdown();
300+
});
301+
302+
verify(streamingCall).sendError(any());
303+
verify(xdsTransport).shutdown();
304+
verify(responseHandler, never())
305+
.handleResourceResponse(any(), any(), any(), any(), any(), any(Boolean.class), any());
306+
}
307+
276308
private static DiscoveryRequest argThatTypeUrlIs(String typeUrl) {
277309
return argThat(req -> req != null && typeUrl.equals(req.getTypeUrl()));
278310
}

0 commit comments

Comments
 (0)