  void initialize() override {
    if (use_eds_) {
      on_server_init_function_ = [this]() {
        AssertionResult result =
            fake_upstreams_[1]->waitForHttpConnection(*dispatcher_, eds_connection_);
        RELEASE_ASSERT(result, result.message());
        result = eds_connection_->waitForNewStream(*dispatcher_, eds_stream_);
        RELEASE_ASSERT(result, result.message());
        eds_stream_->startGrpcStream();

        envoy::service::discovery::v3::DiscoveryRequest discovery_request;
        result = eds_stream_->waitForGrpcMessage(*dispatcher_, discovery_request);
        RELEASE_ASSERT(result, result.message());

        envoy::service::discovery::v3::DiscoveryResponse discovery_response;
        discovery_response.set_version_info("1");
        discovery_response.set_type_url(Config::TypeUrl::get().ClusterLoadAssignment);

        auto cluster_load_assignment =
            TestUtility::parseYaml<envoy::config::endpoint::v3::ClusterLoadAssignment>(fmt::format(
                R"EOF(
                cluster_name: cluster_0
                endpoints:
                - lb_endpoints:
                  - endpoint:
                      address:
                        socket_address:
                          address: {}
                          port_value: {}
                    metadata:
                      filter_metadata:
                        test.namespace:
                          key: metadata-value
              )EOF",
                Network::Test::getLoopbackAddressString(std::get<0>(GetParam())),
                fake_upstreams_[0]->localAddress()->ip()->port()));

        discovery_response.add_resources()->PackFrom(cluster_load_assignment);
        eds_stream_->sendGrpcMessage(discovery_response);

        // Wait for the next request to make sure the first response was consumed.
        result = eds_stream_->waitForGrpcMessage(*dispatcher_, discovery_request);
        RELEASE_ASSERT(result, result.message());
      };
    }

    HttpIntegrationTest::initialize();
  }
  void initialize() override {
    if (use_eds_) {
      on_server_init_function_ = [this]() {
        AssertionResult result =
            fake_upstreams_[1]->waitForHttpConnection(*dispatcher_, eds_connection_);
        RELEASE_ASSERT(result, result.message());
        result = eds_connection_->waitForNewStream(*dispatcher_, eds_stream_);
        RELEASE_ASSERT(result, result.message());
        eds_stream_->startGrpcStream();

        envoy::service::discovery::v3::DiscoveryRequest discovery_request;
        result = eds_stream_->waitForGrpcMessage(*dispatcher_, discovery_request);
        RELEASE_ASSERT(result, result.message());

        envoy::service::discovery::v3::DiscoveryResponse discovery_response;
        discovery_response.set_version_info("1");
        discovery_response.set_type_url(Config::TypeUrl::get().ClusterLoadAssignment);

        auto cluster_load_assignment =
            TestUtility::parseYaml<envoy::config::endpoint::v3::ClusterLoadAssignment>(fmt::format(
                R"EOF(
                cluster_name: cluster_0
                endpoints:
                - lb_endpoints:
                  - endpoint:
                      address:
                        socket_address:
                          address: {}
                          port_value: {}
                    metadata:
                      filter_metadata:
                        test.namespace:
                          key: metadata-value
              )EOF",
                Network::Test::getLoopbackAddressString(std::get<0>(GetParam())),
                fake_upstreams_[0]->localAddress()->ip()->port()));

        discovery_response.add_resources()->PackFrom(cluster_load_assignment);
        eds_stream_->sendGrpcMessage(discovery_response);

         
        result = eds_stream_->waitForGrpcMessage(*dispatcher_, discovery_request);
        RELEASE_ASSERT(result, result.message());
      };
    }

    HttpIntegrationTest::initialize();
  }