TEST_P(CdsIntegrationTest, CdsClusterDownWithLotsOfIdleConnections) {
  constexpr int num_requests = 2000;
  // Make upstream H/1 so it creates connection for each request
  upstream_codec_type_ = Http::CodecType::HTTP1;
  // Relax default circuit breaker limits and timeouts so Envoy can accumulate a lot of idle
  // connections
  cluster_creator_ = &ConfigHelper::buildH1ClusterWithHighCircuitBreakersLimits;
  config_helper_.addConfigModifier(
      [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
              hcm) -> void {
        hcm.mutable_route_config()
            ->mutable_virtual_hosts(0)
            ->mutable_routes(0)
            ->mutable_route()
            ->mutable_timeout()
            ->set_seconds(600);
        hcm.mutable_route_config()
            ->mutable_virtual_hosts(0)
            ->mutable_routes(0)
            ->mutable_route()
            ->mutable_idle_timeout()
            ->set_seconds(600);
      });
  initialize();
  std::vector<IntegrationStreamDecoderPtr> responses;
  std::vector<FakeHttpConnectionPtr> upstream_connections;
  std::vector<FakeStreamPtr> upstream_requests;
  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
  // The first loop establishes a lot of open connections with active requests to upstream
  for (int i = 0; i < num_requests; ++i) {
    Http::TestRequestHeaderMapImpl request_headers{{":method", "GET"},
                                                   {":path", "/cluster1"},
                                                   {":scheme", "http"},
                                                   {":authority", "host"},
                                                   {"x-lyft-user-id", absl::StrCat(i)}};

    auto response = codec_client_->makeHeaderOnlyRequest(request_headers);
    responses.push_back(std::move(response));

    FakeHttpConnectionPtr fake_upstream_connection;
    waitForNextUpstreamConnection({UpstreamIndex1}, TestUtility::DefaultTimeout,
                                  fake_upstream_connection);
    // Wait for the next stream on the upstream connection.
    FakeStreamPtr upstream_request;
    AssertionResult result =
        fake_upstream_connection->waitForNewStream(*dispatcher_, upstream_request);
    RELEASE_ASSERT(result, result.message());
    // Wait for the stream to be completely received.
    result = upstream_request->waitForEndStream(*dispatcher_);
    RELEASE_ASSERT(result, result.message());
    upstream_connections.push_back(std::move(fake_upstream_connection));
    upstream_requests.push_back(std::move(upstream_request));
  }

  // This loop completes all requests making the all upstream connections idle
  for (int i = 0; i < num_requests; ++i) {
    // Send response headers, and end_stream if there is no response body.
    upstream_requests[i]->encodeHeaders(default_response_headers_, true);
    // Wait for the response to be read by the codec client.
    RELEASE_ASSERT(responses[i]->waitForEndStream(), "unexpected timeout");
    ASSERT_TRUE(responses[i]->complete());
    EXPECT_EQ("200", responses[i]->headers().getStatusValue());
  }

  test_server_->waitForCounterGe("cluster_manager.cluster_added", 1);

  // Tell Envoy that cluster_1 is gone. Envoy will try to close all idle connections
  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Cluster, "55", {}, {}, {}));
  sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(Config::TypeUrl::get().Cluster, {}, {},
                                                             {ClusterName1}, "42");
  // We can continue the test once we're sure that Envoy's ClusterManager has made use of
  // the DiscoveryResponse that says cluster_1 is gone.
  test_server_->waitForCounterGe("cluster_manager.cluster_removed", 1);

  // If we made it this far then everything is ok.
  for (int i = 0; i < num_requests; ++i) {
    AssertionResult result = upstream_connections[i]->close();
    RELEASE_ASSERT(result, result.message());
    result = upstream_connections[i]->waitForDisconnect();
    RELEASE_ASSERT(result, result.message());
  }
  upstream_connections.clear();
  cleanupUpstreamAndDownstream();
  ASSERT_TRUE(codec_client_->waitForDisconnect());
}
TEST_P(CdsIntegrationTest, CdsClusterDownWithLotsOfIdleConnections) {
  constexpr int num_requests = 2000;
   
  upstream_codec_type_ = Http::CodecType::HTTP1;
   
   
  cluster_creator_ = &ConfigHelper::buildH1ClusterWithHighCircuitBreakersLimits;
  config_helper_.addConfigModifier(
      [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
              hcm) -> void {
        hcm.mutable_route_config()
            ->mutable_virtual_hosts(0)
            ->mutable_routes(0)
            ->mutable_route()
            ->mutable_timeout()
            ->set_seconds(600);
        hcm.mutable_route_config()
            ->mutable_virtual_hosts(0)
            ->mutable_routes(0)
            ->mutable_route()
            ->mutable_idle_timeout()
            ->set_seconds(600);
      });
  initialize();
  std::vector<IntegrationStreamDecoderPtr> responses;
  std::vector<FakeHttpConnectionPtr> upstream_connections;
  std::vector<FakeStreamPtr> upstream_requests;
  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
   
  for (int i = 0; i < num_requests; ++i) {
    Http::TestRequestHeaderMapImpl request_headers{{":method", "GET"},
                                                   {":path", "/cluster1"},
                                                   {":scheme", "http"},
                                                   {":authority", "host"},
                                                   {"x-lyft-user-id", absl::StrCat(i)}};

    auto response = codec_client_->makeHeaderOnlyRequest(request_headers);
    responses.push_back(std::move(response));

    FakeHttpConnectionPtr fake_upstream_connection;
    waitForNextUpstreamConnection({UpstreamIndex1}, TestUtility::DefaultTimeout,
                                  fake_upstream_connection);
     
    FakeStreamPtr upstream_request;
    AssertionResult result =
        fake_upstream_connection->waitForNewStream(*dispatcher_, upstream_request);
    RELEASE_ASSERT(result, result.message());
     
    result = upstream_request->waitForEndStream(*dispatcher_);
    RELEASE_ASSERT(result, result.message());
    upstream_connections.push_back(std::move(fake_upstream_connection));
    upstream_requests.push_back(std::move(upstream_request));
  }

   
  for (int i = 0; i < num_requests; ++i) {
     
    upstream_requests[i]->encodeHeaders(default_response_headers_, true);
     
    RELEASE_ASSERT(responses[i]->waitForEndStream(), "unexpected timeout");
    ASSERT_TRUE(responses[i]->complete());
    EXPECT_EQ("200", responses[i]->headers().getStatusValue());
  }

  test_server_->waitForCounterGe("cluster_manager.cluster_added", 1);

   
  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Cluster, "55", {}, {}, {}));
  sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(Config::TypeUrl::get().Cluster, {}, {},
                                                             {ClusterName1}, "42");
   
   
  test_server_->waitForCounterGe("cluster_manager.cluster_removed", 1);

   
  for (int i = 0; i < num_requests; ++i) {
    AssertionResult result = upstream_connections[i]->close();
    RELEASE_ASSERT(result, result.message());
    result = upstream_connections[i]->waitForDisconnect();
    RELEASE_ASSERT(result, result.message());
  }
  upstream_connections.clear();
  cleanupUpstreamAndDownstream();
  ASSERT_TRUE(codec_client_->waitForDisconnect());
}