From 092d3fc9dd3753ef9a2a3f5693ee75bb5d5172f4 Mon Sep 17 00:00:00 2001 From: Ran Date: Tue, 20 Oct 2026 11:08:36 -0700 Subject: [PATCH] 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(). --- .../grpc/xds/client/ControlPlaneClient.java | 28 ++--- .../io/grpc/xds/client/XdsClientImpl.java | 5 + .../io/grpc/xds/XdsClientFallbackTest.java | 111 ++++++++++++++++++ .../xds/client/ControlPlaneClientTest.java | 32 +++++ 4 files changed, 162 insertions(+), 14 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java b/xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java index 981db516e5b..75dcaf29ef8 100644 --- a/xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java +++ b/xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java @@ -114,21 +114,21 @@ final class ControlPlaneClient { logger.log(XdsLogLevel.INFO, "Created"); } + // Must be synchronized. void shutdown() { - syncContext.execute(new Runnable() { - @Override - public void run() { - shutdown = true; - logger.log(XdsLogLevel.INFO, "Shutting down"); - if (adsStream != null) { - adsStream.close(Status.CANCELLED.withDescription("shutdown").asException()); - } - if (rpcRetryTimer != null && rpcRetryTimer.isPending()) { - rpcRetryTimer.cancel(); - } - xdsTransport.shutdown(); - } - }); + syncContext.throwIfNotInThisSynchronizationContext(); + if (shutdown) { + return; + } + shutdown = true; + logger.log(XdsLogLevel.INFO, "Shutting down"); + if (adsStream != null) { + adsStream.close(Status.CANCELLED.withDescription("shutdown").asException()); + } + if (rpcRetryTimer != null && rpcRetryTimer.isPending()) { + rpcRetryTimer.cancel(); + } + xdsTransport.shutdown(); } @Override diff --git a/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java b/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java index 7e9f712db17..b9b8f9687af 100644 --- a/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java +++ b/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java @@ -574,6 +574,11 @@ private void handleResourceUpdate( XdsResourceType.Args args, List resources, XdsResourceType xdsResourceType, boolean isFirstResponse, ProcessingTracker processingTracker) { ControlPlaneClient controlPlaneClient = serverCpClientMap.get(args.serverInfo); + if (controlPlaneClient == null) { + logger.log(XdsLogLevel.DEBUG, + "Couldn't find CPC for {0}, so skipping resource update", args.serverInfo); + return; + } if (isFirstResponse) { shutdownLowerPriorityCpcs(controlPlaneClient); diff --git a/xds/src/test/java/io/grpc/xds/XdsClientFallbackTest.java b/xds/src/test/java/io/grpc/xds/XdsClientFallbackTest.java index c414e7e6de0..493d2d04c2c 100644 --- a/xds/src/test/java/io/grpc/xds/XdsClientFallbackTest.java +++ b/xds/src/test/java/io/grpc/xds/XdsClientFallbackTest.java @@ -22,15 +22,20 @@ import static org.mockito.AdditionalAnswers.delegatesTo; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.protobuf.Any; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; import io.grpc.ChannelCredentials; import io.grpc.Grpc; import io.grpc.MetricRecorder; @@ -591,6 +596,112 @@ public void used_then_mainServerRestart_fallbackServerUp() { assertThat(getLrsServerInfo("localhost:" + fallbackServer.getServer().getPort())).isNull(); } + @Test + public void concurrentResponsesOnMainAndFallback_shutsDownFallbackWithoutNpe() { + @SuppressWarnings("unchecked") + XdsTransportFactory.StreamingCall mainCall1 = + mock(XdsTransportFactory.StreamingCall.class); + @SuppressWarnings("unchecked") + XdsTransportFactory.StreamingCall mainCall2 = + mock(XdsTransportFactory.StreamingCall.class); + @SuppressWarnings("unchecked") + XdsTransportFactory.StreamingCall fallbackCall = + mock(XdsTransportFactory.StreamingCall.class); + when(mainCall1.isReady()).thenReturn(true); + when(mainCall2.isReady()).thenReturn(true); + when(fallbackCall.isReady()).thenReturn(true); + + XdsTransportFactory.XdsTransport mainTransport = mock(XdsTransportFactory.XdsTransport.class); + XdsTransportFactory.XdsTransport fallbackTransport = + mock(XdsTransportFactory.XdsTransport.class); + when(mainTransport.createStreamingCall( + any(), any(), any())) + .thenReturn(mainCall1) + .thenReturn(mainCall2); + when(fallbackTransport.createStreamingCall( + any(), any(), any())) + .thenReturn(fallbackCall); + + XdsTransportFactory customTransportFactory = + serverInfo -> + serverInfo.target().equals("main-uri") ? mainTransport : fallbackTransport; + + XdsClientImpl client = + CommonBootstrapperTestUtils.createXdsClient( + Arrays.asList("main-uri", "fallback-uri"), + customTransportFactory, + fakeClock, + new ExponentialBackoffPolicy.Provider(), + MessagePrinter.INSTANCE, + xdsClientMetricReporter); + + client.watchXdsResource(XdsListenerResource.getInstance(), MAIN_SERVER, ldsWatcher); + + @SuppressWarnings("unchecked") + ArgumentCaptor> mainHandler1Captor = + ArgumentCaptor.forClass(XdsTransportFactory.EventHandler.class); + verify(mainCall1).start(mainHandler1Captor.capture()); + + // Fail main's initial ADS stream before any response, activating fallback. + mainHandler1Captor.getValue().onStatusReceived(Status.UNAVAILABLE); + + @SuppressWarnings("unchecked") + ArgumentCaptor> fallbackHandlerCaptor = + ArgumentCaptor.forClass(XdsTransportFactory.EventHandler.class); + verify(fallbackCall).start(fallbackHandlerCaptor.capture()); + fallbackHandlerCaptor.getValue().onReady(); + + // Advance clock so main's retry timer fires and opens a second ADS stream while fallback's + // stream is also open. + fakeClock.forwardTime(2, TimeUnit.SECONDS); + + @SuppressWarnings("unchecked") + ArgumentCaptor> mainHandler2Captor = + ArgumentCaptor.forClass(XdsTransportFactory.EventHandler.class); + verify(mainCall2).start(mainHandler2Captor.capture()); + + DiscoveryResponse mainResponse = + DiscoveryResponse.newBuilder() + .setTypeUrl(ADS_TYPE_URL_LDS) + .setVersionInfo("1") + .setNonce("nonce-main") + .addResources(Any.pack(ControlPlaneRule.buildClientListener(MAIN_SERVER, RDS_NAME))) + .build(); + DiscoveryResponse fallbackResponse = + DiscoveryResponse.newBuilder() + .setTypeUrl(ADS_TYPE_URL_LDS) + .setVersionInfo("1") + .setNonce("nonce-fallback") + .addResources( + Any.pack(ControlPlaneRule.buildClientListener(MAIN_SERVER, FALLBACK_RDS_NAME))) + .build(); + + // When mainCall2 sends its initial DiscoveryRequest (inside XdsClientImpl's syncContext), + // enqueue onRecvMessage for both main and fallback before syncContext finishes draining. + // Main's first response runs shutdownLowerPriorityCpcs(mainCpc), and fallback's already-queued + // onRecvMessage runs immediately after in the same drain loop. + doAnswer( + invocation -> { + mainHandler2Captor.getValue().onRecvMessage(mainResponse); + fallbackHandlerCaptor.getValue().onRecvMessage(fallbackResponse); + return null; + }) + .doNothing() + .when(mainCall2) + .sendMessage(any()); + + mainHandler2Captor.getValue().onReady(); + + verify(ldsWatcher, timeout(5000)) + .onResourceChanged( + StatusOr.fromValue(LdsUpdate.forApiListener(MAIN_HTTP_CONNECTION_MANAGER))); + verify(fallbackCall).sendError(any()); + verify(fallbackTransport).shutdown(); + verify(fallbackCall, times(1)).sendMessage(any()); + + client.shutdown(); + } + /** * gRFC A95: for an authority with fallback_on_reachability_only, falling back must happen when * the primary server becomes unreachable, even though every subscribed resource is cached. diff --git a/xds/src/test/java/io/grpc/xds/client/ControlPlaneClientTest.java b/xds/src/test/java/io/grpc/xds/client/ControlPlaneClientTest.java index 64786c4fb3b..0dc0c7edae0 100644 --- a/xds/src/test/java/io/grpc/xds/client/ControlPlaneClientTest.java +++ b/xds/src/test/java/io/grpc/xds/client/ControlPlaneClientTest.java @@ -273,6 +273,38 @@ public void cancelBeforeAck_sendsEmptyUnsubscribe() { assertThat(unsub.getResourceNamesList()).isEmpty(); } + @Test + public void shutdown_closesStreamImmediatelyAndSkipsQueuedRecvMessage() { + Map> subscribedTypes = ImmutableMap.of(CDS_TYPE_URL, cdsType); + when(resourceStore.getSubscribedResourceTypesWithTypeUrl()).thenReturn(subscribedTypes); + when(resourceStore.getSubscribedResources(serverInfo, cdsType)) + .thenReturn(ImmutableList.of("foo-cluster")); + + syncContext.execute(cpc::sendDiscoveryRequests); + verify(streamingCall).start(handlerCaptor.capture()); + handlerCaptor.getValue().onReady(); + + DiscoveryResponse response = + DiscoveryResponse.newBuilder() + .setTypeUrl(CDS_TYPE_URL) + .setVersionInfo("1") + .setNonce("nonce-1") + .build(); + + // Simulate a response arriving while syncContext is draining a task that shuts down this CPC + // (e.g. XdsClientImpl.shutdownLowerPriorityCpcs when a higher-priority server responds first). + syncContext.execute( + () -> { + handlerCaptor.getValue().onRecvMessage(response); + cpc.shutdown(); + }); + + verify(streamingCall).sendError(any()); + verify(xdsTransport).shutdown(); + verify(responseHandler, never()) + .handleResourceResponse(any(), any(), any(), any(), any(), any(Boolean.class), any()); + } + private static DiscoveryRequest argThatTypeUrlIs(String typeUrl) { return argThat(req -> req != null && typeUrl.equals(req.getTypeUrl())); }