TEST_F(RouterTest, HedgedPerTryTimeoutThirdRequestSucceeds) {
  enableHedgeOnPerTryTimeout();

  // Verify cluster request/response sizes are accounted for all requests/responses.
  EXPECT_CALL(
      cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_headers_size"), 73ull))
      .Times(3);
  EXPECT_CALL(
      cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_body_size"), 0ull))
      .Times(3);
  EXPECT_CALL(
      cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_headers_size"), 10ull))
      .Times(2);
  EXPECT_CALL(
      cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_body_size"), 0ull))
      .Times(2);

  NiceMock<Http::MockRequestEncoder> encoder1;
  Http::ResponseDecoder* response_decoder1 = nullptr;
  EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_, newStream(_, _))
      .WillOnce(Invoke(
          [&](Http::ResponseDecoder& decoder,
              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
            response_decoder1 = &decoder;
            EXPECT_CALL(*router_.retry_state_, onHostAttempted(_));
            upstream_stream_info_.downstream_connection_info_provider_->setConnectionID(111);
            callbacks.onPoolReady(encoder1, cm_.thread_local_cluster_.conn_pool_.host_,
                                  upstream_stream_info_, Http::Protocol::Http10);
            return nullptr;
          }));
  expectResponseTimerCreate();
  expectPerTryTimerCreate();

  Http::TestRequestHeaderMapImpl headers{{"x-envoy-upstream-rq-per-try-timeout-ms", "5"}};
  HttpTestUtility::addDefaultHeaders(headers);
  router_.decodeHeaders(headers, true);
  EXPECT_EQ(1U,
            callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());

  EXPECT_CALL(encoder1.stream_, resetStream(_)).Times(0);

  Http::ResponseHeaderMapPtr response_headers1(
      new Http::TestResponseHeaderMapImpl{{":status", "500"}});
  // Local origin connect success happens for first and third try.
  EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_.host_->outlier_detector_,
              putResult(Upstream::Outlier::Result::LocalOriginConnectSuccess,
                        absl::optional<uint64_t>(absl::nullopt)))
      .Times(2);
  EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_.host_->outlier_detector_,
              putHttpResponseCode(500));
  EXPECT_CALL(encoder1.stream_, resetStream(_)).Times(0);
  EXPECT_CALL(callbacks_, encodeHeaders_(_, _)).Times(0);
  router_.retry_state_->expectHeadersRetry();
  // NOLINTNEXTLINE: Silence null pointer access warning
  response_decoder1->decodeHeaders(std::move(response_headers1), true);

  NiceMock<Http::MockRequestEncoder> encoder2;
  Http::ResponseDecoder* response_decoder2 = nullptr;
  EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_, newStream(_, _))
      .WillOnce(Invoke(
          [&](Http::ResponseDecoder& decoder,
              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
            response_decoder2 = &decoder;
            EXPECT_CALL(*router_.retry_state_, onHostAttempted(_));
            upstream_stream_info_.downstream_connection_info_provider_->setConnectionID(222);
            callbacks.onPoolReady(encoder2, cm_.thread_local_cluster_.conn_pool_.host_,
                                  upstream_stream_info_, Http::Protocol::Http10);
            return nullptr;
          }));
  expectPerTryTimerCreate();
  router_.retry_state_->callback_();
  EXPECT_EQ(2U,
            callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());

  EXPECT_TRUE(verifyHostUpstreamStats(0, 1));

  // Now trigger a per try timeout on the 2nd request, expect a 3rd
  router_.retry_state_->expectHedgedPerTryTimeoutRetry();
  EXPECT_CALL(
      cm_.thread_local_cluster_.conn_pool_.host_->outlier_detector_,
      putResult(Upstream::Outlier::Result::LocalOriginTimeout, absl::optional<uint64_t>(504)));
  NiceMock<Http::MockRequestEncoder> encoder3;
  Http::ResponseDecoder* response_decoder3 = nullptr;
  EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_, newStream(_, _))
      .WillOnce(Invoke(
          [&](Http::ResponseDecoder& decoder,
              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
            response_decoder3 = &decoder;
            EXPECT_CALL(*router_.retry_state_, onHostAttempted(_));
            upstream_stream_info_.downstream_connection_info_provider_->setConnectionID(333);
            callbacks.onPoolReady(encoder3, cm_.thread_local_cluster_.conn_pool_.host_,
                                  upstream_stream_info_, Http::Protocol::Http10);
            return nullptr;
          }));

  EXPECT_CALL(callbacks_, encodeHeaders_(_, _)).Times(0);
  per_try_timeout_->invokeCallback();
  expectPerTryTimerCreate();
  router_.retry_state_->callback_();
  EXPECT_EQ(3U,
            callbacks_.route_->route_entry_.virtual_cluster_.stats().upstream_rq_total_.value());
  EXPECT_TRUE(verifyHostUpstreamStats(0, 1));

  // Now write a 200 back. We expect the 2nd stream to be reset and stats to be
  // incremented properly.
  Http::ResponseHeaderMapPtr response_headers2(
      new Http::TestResponseHeaderMapImpl{{":status", "200"}});
  EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_.host_->outlier_detector_,
              putHttpResponseCode(200));
  EXPECT_CALL(encoder1.stream_, resetStream(_)).Times(0);
  EXPECT_CALL(encoder2.stream_, resetStream(_));
  EXPECT_CALL(encoder3.stream_, resetStream(_)).Times(0);

  EXPECT_CALL(callbacks_, encodeHeaders_(_, _))
      .WillOnce(Invoke([&](Http::ResponseHeaderMap& headers, bool end_stream) -> void {
        EXPECT_EQ(headers.Status()->value(), "200");
        EXPECT_TRUE(end_stream);
      }));
  EXPECT_CALL(*router_.retry_state_, shouldRetryHeaders(_, _)).WillOnce(Return(RetryStatus::No));
  // NOLINTNEXTLINE: Silence null pointer access warning
  response_decoder3->decodeHeaders(std::move(response_headers2), true);
  EXPECT_TRUE(verifyHostUpstreamStats(1, 1));

  EXPECT_EQ(333U, callbacks_.stream_info_.upstreamInfo()->upstreamConnectionId());

  // TODO: Verify hedge stats here once they are implemented.
}