xds_cluster_type_end2end_test: fix flake in AggregateClusterTest.FallBackWithConnectivityChurn test (#29786)

src/core/lib/surface/server.cc

@@ -829,8 +829,13 @@ void Server::ShutdownAndNotify(grpc_completion_queue* cq, void* tag) {
   if (await_requests != nullptr) {
     await_requests->WaitForNotification();
   }
-  // Shutdown listeners.
+  StopListening();
+  broadcaster.BroadcastShutdown(/*send_goaway=*/true, GRPC_ERROR_NONE);
+}
+
+void Server::StopListening() {
   for (auto& listener : listeners_) {
+    if (listener.listener == nullptr) continue;
     channelz::ListenSocketNode* channelz_listen_socket_node =
         listener.listener->channelz_listen_socket_node();
     if (channelz_node_ != nullptr && channelz_listen_socket_node != nullptr) {
@@ -842,7 +847,6 @@ void Server::ShutdownAndNotify(grpc_completion_queue* cq, void* tag) {
     listener.listener->SetOnDestroyDone(&listener.destroy_done);
     listener.listener.reset();
   }
-  broadcaster.BroadcastShutdown(/*send_goaway=*/true, GRPC_ERROR_NONE);
 }
 
 void Server::CancelAllCalls() {

src/core/lib/surface/server.h

@@ -195,6 +195,8 @@ class Server : public InternallyRefCounted<Server>,
   void ShutdownAndNotify(grpc_completion_queue* cq, void* tag)
       ABSL_LOCKS_EXCLUDED(mu_global_, mu_call_);
 
+  void StopListening();
+
   void CancelAllCalls() ABSL_LOCKS_EXCLUDED(mu_global_);
 
   void SendGoaways() ABSL_LOCKS_EXCLUDED(mu_global_, mu_call_);

test/cpp/end2end/xds/xds_cluster_type_end2end_test.cc

@@ -524,8 +524,11 @@ TEST_P(AggregateClusterTest, FallBackWithConnectivityChurn) {
   // Increase timeout to account for subchannel connection delays.
   WaitForBackend(DEBUG_LOCATION, 0, WaitForBackendOptions(),
                  RpcOptions().set_timeout_ms(2000));
-  // Bring down the P0 backend.
-  ShutdownBackend(0);
+  // Send GOAWAY from the P0 backend.
+  // We don't actually shut it down here to avoid flakiness caused by
+  // failing an RPC after the client has already sent it but before the
+  // server finished processing it.
+  backends_[0]->StopListeningAndSendGoaways();
   // Allow the connection attempt to the P1 backend to resume.
   connection_attempt_injector.CompletePriority1Connection();
   // Wait for P1 backend to start getting traffic.

test/cpp/end2end/xds/xds_end2end_test_lib.cc

@@ -41,6 +41,7 @@
 #include "src/core/ext/xds/xds_client.h"
 #include "src/core/lib/gpr/env.h"
 #include "src/core/lib/gpr/tmpfile.h"
+#include "src/core/lib/surface/server.h"
 #include "src/cpp/client/secure_credentials.h"
 #include "src/proto/grpc/testing/xds/v3/router.grpc.pb.h"
 #include "test/core/util/resolve_localhost_ip46.h"
@@ -159,6 +160,17 @@ void XdsEnd2endTest::ServerThread::Shutdown() {
   running_ = false;
 }
 
+void XdsEnd2endTest::ServerThread::StopListeningAndSendGoaways() {
+  gpr_log(GPR_INFO, "%s sending GOAWAYs", Type());
+  {
+    grpc_core::ExecCtx exec_ctx;
+    auto* server = grpc_core::Server::FromC(server_->c_server());
+    server->StopListening();
+    server->SendGoaways();
+  }
+  gpr_log(GPR_INFO, "%s done sending GOAWAYs", Type());
+}
+
 void XdsEnd2endTest::ServerThread::Serve(grpc_core::Mutex* mu,
                                          grpc_core::CondVar* cond) {
   // We need to acquire the lock here in order to prevent the notify_one

test/cpp/end2end/xds/xds_end2end_test_lib.h

@@ -257,6 +257,8 @@ class XdsEnd2endTest : public ::testing::TestWithParam<XdsTestType> {
       allow_put_requests_ = allow_put_requests;
     }
 
+    void StopListeningAndSendGoaways();
+
    private:
     class XdsChannelArgsServerBuilderOption;