Add a test case to verify SubchannelStreamClient retry when Health.Watch ends (#31850)

<!--

If you know who should review your pull request, please assign it to
that
person, otherwise the pull request would get assigned randomly.

If your pull request is for a specific language, please add the
appropriate
lang label.

-->

CMakeLists.txt

@@ -8798,6 +8798,10 @@ if(gRPC_BUILD_TESTS)
 if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 
   add_executable(client_lb_end2end_test
+    ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/health/v1/health.pb.cc
+    ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/health/v1/health.grpc.pb.cc
+    ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/health/v1/health.pb.h
+    ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/health/v1/health.grpc.pb.h
     ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc
     ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc
     ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/duplicate/echo_duplicate.pb.h

build_autogenerated.yaml

@@ -5999,6 +5999,7 @@ targets:
   - test/cpp/end2end/connection_attempt_injector.h
   - test/cpp/end2end/test_service_impl.h
   src:
+  - src/proto/grpc/health/v1/health.proto
   - src/proto/grpc/testing/duplicate/echo_duplicate.proto
   - src/proto/grpc/testing/echo.proto
   - src/proto/grpc/testing/echo_messages.proto

test/cpp/end2end/BUILD

@@ -499,6 +499,7 @@ grpc_cc_test(
         "//:grpcpp_call_metric_recorder",
         "//:grpcpp_orca_service",
         "//src/core:channel_args",
+        "//src/proto/grpc/health/v1:health_proto",
         "//src/proto/grpc/testing:echo_messages_proto",
         "//src/proto/grpc/testing:echo_proto",
         "//src/proto/grpc/testing/duplicate:echo_duplicate_proto",

test/cpp/end2end/client_lb_end2end_test.cc

@@ -69,6 +69,7 @@
 #include "src/core/lib/transport/connectivity_state.h"
 #include "src/cpp/client/secure_credentials.h"
 #include "src/cpp/server/secure_server_credentials.h"
+#include "src/proto/grpc/health/v1/health.grpc.pb.h"
 #include "src/proto/grpc/testing/echo.grpc.pb.h"
 #include "src/proto/grpc/testing/xds/v3/orca_load_report.pb.h"
 #include "test/core/util/port.h"
@@ -84,6 +85,32 @@ namespace {
 
 constexpr char kRequestMessage[] = "Live long and prosper.";
 
+// A noop health check service that just terminates the call and returns OK
+// status in its methods. This is used to test the retry mechanism in
+// SubchannelStreamClient.
+class NoopHealthCheckServiceImpl : public health::v1::Health::Service {
+ public:
+  ~NoopHealthCheckServiceImpl() override = default;
+  Status Check(ServerContext*, const health::v1::HealthCheckRequest*,
+               health::v1::HealthCheckResponse*) override {
+    return Status::OK;
+  }
+  Status Watch(ServerContext*, const health::v1::HealthCheckRequest*,
+               ServerWriter<health::v1::HealthCheckResponse>*) override {
+    grpc_core::MutexLock lock(&mu_);
+    request_count_++;
+    return Status::OK;
+  }
+  int request_count() {
+    grpc_core::MutexLock lock(&mu_);
+    return request_count_;
+  }
+
+ private:
+  grpc_core::Mutex mu_;
+  int request_count_ ABSL_GUARDED_BY(&mu_) = 0;
+};
+
 // Subclass of TestServiceImpl that increments a request counter for
 // every call to the Echo RPC.
 class MyTestServiceImpl : public TestServiceImpl {
@@ -405,6 +432,8 @@ class ClientLbEnd2endTest : public ::testing::Test {
     std::unique_ptr<experimental::ServerMetricRecorder> server_metric_recorder_;
     experimental::OrcaService orca_service_;
     std::unique_ptr<std::thread> thread_;
+    bool enable_noop_health_check_service_ = false;
+    NoopHealthCheckServiceImpl noop_health_check_service_impl_;
 
     grpc_core::Mutex mu_;
     grpc_core::CondVar cond_;
@@ -441,6 +470,9 @@ class ClientLbEnd2endTest : public ::testing::Test {
       builder.AddListeningPort(server_address.str(), std::move(creds));
       builder.RegisterService(&service_);
       builder.RegisterService(&orca_service_);
+      if (enable_noop_health_check_service_) {
+        builder.RegisterService(&noop_health_check_service_impl_);
+      }
       grpc::ServerBuilder::experimental_type(&builder)
           .EnableCallMetricRecording(server_metric_recorder_.get());
       server_ = builder.BuildAndStart();
@@ -567,6 +599,12 @@ class ClientLbEnd2endTest : public ::testing::Test {
     }
   }
 
+  void EnableNoopHealthCheckService() {
+    for (auto& server : servers_) {
+      server->enable_noop_health_check_service_ = true;
+    }
+  }
+
   static std::string MakeConnectionFailureRegex(absl::string_view prefix) {
     return absl::StrCat(prefix,
                         "; last error: (UNKNOWN|UNAVAILABLE): "
@@ -2187,6 +2225,26 @@ TEST_F(RoundRobinTest,
   EnableDefaultHealthCheckService(false);
 }
 
+TEST_F(RoundRobinTest, HealthCheckingRetryOnStreamEnd) {
+  // Start servers.
+  const int kNumServers = 2;
+  CreateServers(kNumServers);
+  EnableNoopHealthCheckService();
+  StartServer(0);
+  StartServer(1);
+  ChannelArguments args;
+  // Create a channel with health-checking enabled.
+  args.SetServiceConfigJSON(
+      "{\"healthCheckConfig\": "
+      "{\"serviceName\": \"health_check_service_name\"}}");
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("round_robin", response_generator, args);
+  response_generator.SetNextResolution(GetServersPorts());
+  EXPECT_FALSE(WaitForChannelReady(channel.get()));
+  EXPECT_GT(servers_[0]->noop_health_check_service_impl_.request_count(), 1);
+  EXPECT_GT(servers_[1]->noop_health_check_service_impl_.request_count(), 1);
+}
+
 //
 // LB policy pick args
 //