void FilterManager::decodeHeaders(ActiveStreamDecoderFilter* filter, RequestHeaderMap& headers,
                                  bool end_stream) {
  // Headers filter iteration should always start with the next filter if available.
  std::list<ActiveStreamDecoderFilterPtr>::iterator entry =
      commonDecodePrefix(filter, FilterIterationStartState::AlwaysStartFromNext);
  std::list<ActiveStreamDecoderFilterPtr>::iterator continue_data_entry = decoder_filters_.end();

  for (; entry != decoder_filters_.end(); entry++) {
    (*entry)->maybeEvaluateMatchTreeWithNewData(
        [&](auto& matching_data) { matching_data.onRequestHeaders(headers); });

    if ((*entry)->skipFilter()) {
      continue;
    }

    ASSERT(!(state_.filter_call_state_ & FilterCallState::DecodeHeaders));
    state_.filter_call_state_ |= FilterCallState::DecodeHeaders;
    (*entry)->end_stream_ = (end_stream && continue_data_entry == decoder_filters_.end());
    FilterHeadersStatus status = (*entry)->decodeHeaders(headers, (*entry)->end_stream_);
    if (state_.decoder_filter_chain_aborted_) {
      ENVOY_STREAM_LOG(trace,
                       "decodeHeaders filter iteration aborted due to local reply: filter={}",
                       *this, static_cast<const void*>((*entry).get()));
      status = FilterHeadersStatus::StopIteration;
    }

    ASSERT(!(status == FilterHeadersStatus::ContinueAndDontEndStream && !(*entry)->end_stream_),
           "Filters should not return FilterHeadersStatus::ContinueAndDontEndStream from "
           "decodeHeaders when end_stream is already false");

    state_.filter_call_state_ &= ~FilterCallState::DecodeHeaders;
    ENVOY_STREAM_LOG(trace, "decode headers called: filter={} status={}", *this,
                     static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status));

    (*entry)->decode_headers_called_ = true;

    const auto continue_iteration = (*entry)->commonHandleAfterHeadersCallback(status, end_stream);
    ENVOY_BUG(!continue_iteration || !state_.local_complete_,
              "Filter did not return StopAll or StopIteration after sending a local reply.");

    // If this filter ended the stream, decodeComplete() should be called for it.
    if ((*entry)->end_stream_) {
      (*entry)->handle_->decodeComplete();
    }

    // Skip processing metadata after sending local reply
    if (state_.local_complete_ && std::next(entry) != decoder_filters_.end()) {
      maybeContinueDecoding(continue_data_entry);
      return;
    }

    const bool new_metadata_added = processNewlyAddedMetadata();
    // If end_stream is set in headers, and a filter adds new metadata, we need to delay end_stream
    // in headers by inserting an empty data frame with end_stream set. The empty data frame is sent
    // after the new metadata.
    if ((*entry)->end_stream_ && new_metadata_added && !buffered_request_data_) {
      Buffer::OwnedImpl empty_data("");
      ENVOY_STREAM_LOG(
          trace, "inserting an empty data frame for end_stream due metadata being added.", *this);
      // Metadata frame doesn't carry end of stream bit. We need an empty data frame to end the
      // stream.
      addDecodedData(*((*entry).get()), empty_data, true);
    }

    if (!continue_iteration && std::next(entry) != decoder_filters_.end()) {
      // Stop iteration IFF this is not the last filter. If it is the last filter, continue with
      // processing since we need to handle the case where a terminal filter wants to buffer, but
      // a previous filter has added body.
      maybeContinueDecoding(continue_data_entry);
      return;
    }

    // Here we handle the case where we have a header only request, but a filter adds a body
    // to it. We need to not raise end_stream = true to further filters during inline iteration.
    if (end_stream && buffered_request_data_ && continue_data_entry == decoder_filters_.end()) {
      continue_data_entry = entry;
    }
  }

  maybeContinueDecoding(continue_data_entry);

  if (end_stream) {
    disarmRequestTimeout();
  }
}