TEST_P(ProtocolIntegrationTest, Retry) {
  config_helper_.addConfigModifier([&](envoy::config::bootstrap::v3::Bootstrap& bootstrap) -> void {
    RELEASE_ASSERT(bootstrap.mutable_static_resources()->clusters_size() == 1, "");
    auto& cluster = *bootstrap.mutable_static_resources()->mutable_clusters(0);
    cluster.mutable_track_cluster_stats()->set_request_response_sizes(true);
  });
  useAccessLog("%UPSTREAM_WIRE_BYTES_SENT% %UPSTREAM_WIRE_BYTES_RECEIVED% "
               "%UPSTREAM_HEADER_BYTES_SENT% %UPSTREAM_HEADER_BYTES_RECEIVED%");
  initialize();
  codec_client_ = makeHttpConnection(lookupPort("http"));
  auto response = codec_client_->makeRequestWithBody(
      Http::TestRequestHeaderMapImpl{{":method", "POST"},
                                     {":path", "/test/long/url"},
                                     {":scheme", "http"},
                                     {":authority", "host"},
                                     {"x-forwarded-for", "10.0.0.1"},
                                     {"x-envoy-retry-on", "5xx"}},
      1024);
  waitForNextUpstreamRequest();
  upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "503"}}, false);

  if (fake_upstreams_[0]->httpType() == Http::CodecType::HTTP1) {
    ASSERT_TRUE(fake_upstream_connection_->waitForDisconnect());
    ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_,
                                                          std::chrono::milliseconds(500)));
  } else {
    ASSERT_TRUE(upstream_request_->waitForReset());
  }
  waitForNextUpstreamRequest();
  upstream_request_->encodeHeaders(default_response_headers_, false);
  upstream_request_->encodeData(512, true);

  ASSERT_TRUE(response->waitForEndStream());
  EXPECT_TRUE(upstream_request_->complete());
  EXPECT_EQ(1024U, upstream_request_->bodyLength());

  EXPECT_TRUE(response->complete());
  EXPECT_EQ("200", response->headers().getStatusValue());
  EXPECT_EQ(512U, response->body().size());
  Stats::Store& stats = test_server_->server().stats();
  if (upstreamProtocol() == Http::CodecType::HTTP2) {
    Stats::CounterSharedPtr counter = TestUtility::findCounter(
        stats, absl::StrCat("cluster.cluster_0.", upstreamProtocolStatsRoot(), ".tx_reset"));
    ASSERT_NE(nullptr, counter);
    EXPECT_EQ(1L, counter->value());
  }
  EXPECT_NE(nullptr,
            test_server_->counter(absl::StrCat("cluster.cluster_0.", upstreamProtocolStatsRoot(),
                                               ".dropped_headers_with_underscores")));

  test_server_->waitUntilHistogramHasSamples("cluster.cluster_0.upstream_rq_headers_size");
  test_server_->waitUntilHistogramHasSamples("cluster.cluster_0.upstream_rs_headers_size");

  auto find_histo_sample_count = [&](const std::string& name) -> uint64_t {
    for (auto& histogram : test_server_->histograms()) {
      if (histogram->name() == name) {
        return TestUtility::readSampleCount(test_server_->server().dispatcher(), *histogram);
      }
    }
    return 0;
  };

  EXPECT_EQ(find_histo_sample_count("cluster.cluster_0.upstream_rq_headers_size"), 2);
  EXPECT_EQ(find_histo_sample_count("cluster.cluster_0.upstream_rs_headers_size"), 2);

  // The two requests are sent with https scheme rather than http for QUIC downstream.
  const size_t quic_https_extra_bytes = (downstreamProtocol() == Http::CodecType::HTTP3 ? 2u : 0u);
  expectUpstreamBytesSentAndReceived(
      BytesCountExpectation(2550 + quic_https_extra_bytes, 635, 414 + quic_https_extra_bytes, 54),
      BytesCountExpectation(2262, 548, 184, 27), BytesCountExpectation(2204, 520, 150, 6));
}