void Filter::onUpstreamHeaders(uint64_t response_code, Http::ResponseHeaderMapPtr&& headers,
                               UpstreamRequest& upstream_request, bool end_stream) {
  ENVOY_STREAM_LOG(debug, "upstream headers complete: end_stream={}", *callbacks_, end_stream);

  modify_headers_(*headers);
  // When grpc-status appears in response headers, convert grpc-status to HTTP status code
  // for outlier detection. This does not currently change any stats or logging and does not
  // handle the case when an error grpc-status is sent as a trailer.
  absl::optional<Grpc::Status::GrpcStatus> grpc_status;
  uint64_t grpc_to_http_status = 0;
  if (grpc_request_) {
    grpc_status = Grpc::Common::getGrpcStatus(*headers);
    if (grpc_status.has_value()) {
      grpc_to_http_status = Grpc::Utility::grpcToHttpStatus(grpc_status.value());
    }
  }

  if (grpc_status.has_value()) {
    upstream_request.upstreamHost()->outlierDetector().putHttpResponseCode(grpc_to_http_status);
  } else {
    upstream_request.upstreamHost()->outlierDetector().putHttpResponseCode(response_code);
  }

  if (headers->EnvoyImmediateHealthCheckFail() != nullptr) {
    upstream_request.upstreamHost()->healthChecker().setUnhealthy(
        Upstream::HealthCheckHostMonitor::UnhealthyType::ImmediateHealthCheckFail);
  }

  bool could_not_retry = false;

  // Check if this upstream request was already retried, for instance after
  // hitting a per try timeout. Don't retry it if we already have.
  if (retry_state_) {
    if (upstream_request.retried()) {
      // We already retried this request (presumably for a per try timeout) so
      // we definitely won't retry it again. Check if we would have retried it
      // if we could.
      could_not_retry = retry_state_->wouldRetryFromHeaders(*headers);
    } else {
      const RetryStatus retry_status =
          retry_state_->shouldRetryHeaders(*headers, [this]() -> void { doRetry(); });
      if (retry_status == RetryStatus::Yes) {
        runRetryOptionsPredicates(upstream_request);
        pending_retries_++;
        upstream_request.upstreamHost()->stats().rq_error_.inc();
        Http::CodeStats& code_stats = httpContext().codeStats();
        code_stats.chargeBasicResponseStat(
            cluster_->statsScope(), config_.stats_.stat_names_.retry_,
            static_cast<Http::Code>(response_code), exclude_http_code_stats_);

        if (!end_stream || !upstream_request.encodeComplete()) {
          upstream_request.resetStream();
        }
        upstream_request.removeFromList(upstream_requests_);

        return;
      } else if (retry_status == RetryStatus::NoOverflow) {
        callbacks_->streamInfo().setResponseFlag(StreamInfo::ResponseFlag::UpstreamOverflow);
        could_not_retry = true;
      } else if (retry_status == RetryStatus::NoRetryLimitExceeded) {
        callbacks_->streamInfo().setResponseFlag(
            StreamInfo::ResponseFlag::UpstreamRetryLimitExceeded);
        could_not_retry = true;
      }
    }
  }

  if (route_entry_->internalRedirectPolicy().enabled() &&
      route_entry_->internalRedirectPolicy().shouldRedirectForResponseCode(
          static_cast<Http::Code>(response_code)) &&
      setupRedirect(*headers)) {
    return;
    // If the redirect could not be handled, fail open and let it pass to the
    // next downstream.
  }

  // Check if we got a "bad" response, but there are still upstream requests in
  // flight awaiting headers or scheduled retries. If so, exit to give them a
  // chance to return before returning a response downstream.
  if (could_not_retry && (numRequestsAwaitingHeaders() > 0 || pending_retries_ > 0)) {
    upstream_request.upstreamHost()->stats().rq_error_.inc();

    // Reset the stream because there are other in-flight requests that we'll
    // wait around for and we're not interested in consuming any body/trailers.
    upstream_request.removeFromList(upstream_requests_)->resetStream();
    return;
  }

  // Make sure any retry timers are destroyed since we may not call cleanup() if end_stream is
  // false.
  if (retry_state_) {
    retry_state_.reset();
  }

  // Only send upstream service time if we received the complete request and this is not a
  // premature response.
  if (DateUtil::timePointValid(downstream_request_complete_time_)) {
    Event::Dispatcher& dispatcher = callbacks_->dispatcher();
    MonotonicTime response_received_time = dispatcher.timeSource().monotonicTime();
    std::chrono::milliseconds ms = std::chrono::duration_cast<std::chrono::milliseconds>(
        response_received_time - downstream_request_complete_time_);
    if (!config_.suppress_envoy_headers_) {
      headers->setEnvoyUpstreamServiceTime(ms.count());
    }
  }

  upstream_request.upstreamCanary(
      (headers->EnvoyUpstreamCanary() && headers->EnvoyUpstreamCanary()->value() == "true") ||
      upstream_request.upstreamHost()->canary());
  chargeUpstreamCode(response_code, *headers, upstream_request.upstreamHost(), false);
  if (!Http::CodeUtility::is5xx(response_code)) {
    handleNon5xxResponseHeaders(grpc_status, upstream_request, end_stream, grpc_to_http_status);
  }

  // Append routing cookies
  for (const auto& header_value : downstream_set_cookies_) {
    headers->addReferenceKey(Http::Headers::get().SetCookie, header_value);
  }

  // TODO(zuercher): If access to response_headers_to_add (at any level) is ever needed outside
  // Router::Filter we'll need to find a better location for this work. One possibility is to
  // provide finalizeResponseHeaders functions on the Router::Config and VirtualHost interfaces.
  route_entry_->finalizeResponseHeaders(*headers, callbacks_->streamInfo());

  downstream_response_started_ = true;
  final_upstream_request_ = &upstream_request;
  // In upstream request hedging scenarios the upstream connection ID set in onPoolReady might not
  // be the connection ID of the upstream connection that ended up receiving upstream headers. Thus
  // reset the upstream connection ID here with the ID of the connection that ultimately was the
  // transport for the final upstream request.
  if (final_upstream_request_->streamInfo().upstreamConnectionId().has_value()) {
    callbacks_->streamInfo().setUpstreamConnectionId(
        final_upstream_request_->streamInfo().upstreamConnectionId().value());
  }
  resetOtherUpstreams(upstream_request);
  if (end_stream) {
    onUpstreamComplete(upstream_request);
  }

  callbacks_->encodeHeaders(std::move(headers), end_stream,
                            StreamInfo::ResponseCodeDetails::get().ViaUpstream);
}