diff --git a/include/envoy/stream_info/filter_state.h b/include/envoy/stream_info/filter_state.h
index db993b59a4d03..486d268b2e7f0 100644
--- a/include/envoy/stream_info/filter_state.h
+++ b/include/envoy/stream_info/filter_state.h
@@ -15,13 +15,36 @@ namespace Envoy {
 namespace StreamInfo {
 
 /**
- * FilterState represents dynamically generated information regarding a
- * stream (TCP or HTTP level) by various filters in Envoy. FilterState can
- * be write-once or write-many.
+ * FilterState represents dynamically generated information regarding a stream (TCP or HTTP level)
+ * or a connection by various filters in Envoy. FilterState can be write-once or write-many.
  */
 class FilterState {
 public:
   enum class StateType { ReadOnly, Mutable };
+  // Objects stored in the FilterState may have different life span. Life span is what controls
+  // how long an object stored in FilterState lives. Implementation of this interface actually
+  // stores objects in a (reverse) tree manner - multiple FilterStateImpl with shorter life span may
+  // share the same FilterStateImpl as parent, which may recursively share parent with other
+  // FilterStateImpl at the same life span. This interface is supposed to be accessed at the leaf
+  // level (FilterChain) for objects with any desired longer life span.
+  //
+  // - FilterChain has the shortest life span, which is as long as the filter chain lives.
+  //
+  // - DownstreamRequest is longer than FilterChain. When internal redirect is enabled, one
+  //   downstream request may create multiple filter chains. DownstreamRequest allows an object to
+  //   survive across filter chains for bookkeeping needs.
+  //
+  // - DownstreamConnection makes an object survive the entire duration of a downstream connection.
+  //   Any stream within this connection can see the same object.
+  //
+  // Note that order matters in this enum because it's assumed that life span grows as enum number
+  // grows.
+  enum LifeSpan {
+    FilterChain,
+    DownstreamRequest,
+    DownstreamConnection,
+    TopSpan = DownstreamConnection
+  };
 
   class Object {
   public:
@@ -41,16 +64,18 @@ class FilterState {
    * @param data_name the name of the data being set.
    * @param data an owning pointer to the data to be stored.
    * @param state_type indicates whether the object is mutable or not.
+   * @param life_span indicates the life span of the object: bound to the filter chain, a
+   * downstream request, or a downstream connection.
    *
    * Note that it is an error to call setData() twice with the same
    * data_name, if the existing object is immutable. Similarly, it is an
    * error to call setData() with same data_name but different state_types
-   * (mutable and readOnly, or readOnly and mutable). This is to enforce a
-   * single authoritative source for each piece of immutable data stored in
-   * FilterState.
+   * (mutable and readOnly, or readOnly and mutable) or different life_span.
+   * This is to enforce a single authoritative source for each piece of
+   * data stored in FilterState.
    */
-  virtual void setData(absl::string_view data_name, std::unique_ptr<Object>&& data,
-                       StateType state_type) PURE;
+  virtual void setData(absl::string_view data_name, std::shared_ptr<Object> data,
+                       StateType state_type, LifeSpan life_span = LifeSpan::FilterChain) PURE;
 
   /**
    * @param data_name the name of the data being looked up (mutable/readonly).
@@ -102,6 +127,24 @@ class FilterState {
    */
   virtual bool hasDataWithName(absl::string_view data_name) const PURE;
 
+  /**
+   * @param life_span the LifeSpan above which data existence is checked.
+   * @return whether data of any type exist with LifeSpan greater than life_span.
+   */
+  virtual bool hasDataAtOrAboveLifeSpan(LifeSpan life_span) const PURE;
+
+  /**
+   * @return the LifeSpan of objects stored by this instance. Objects with
+   * LifeSpan longer than this are handled recursively.
+   */
+  virtual LifeSpan lifeSpan() const PURE;
+
+  /**
+   * @return the pointer of the parent FilterState that has longer life span. nullptr means this is
+   * either the top LifeSpan or the parent is not yet created.
+   */
+  virtual std::shared_ptr<FilterState> parent() const PURE;
+
 protected:
   virtual const Object* getDataReadOnlyGeneric(absl::string_view data_name) const PURE;
   virtual Object* getDataMutableGeneric(absl::string_view data_name) PURE;
diff --git a/source/common/http/conn_manager_impl.cc b/source/common/http/conn_manager_impl.cc
index 3b72848c6eb58..28fe39559a4b9 100644
--- a/source/common/http/conn_manager_impl.cc
+++ b/source/common/http/conn_manager_impl.cc
@@ -14,6 +14,7 @@
 #include "envoy/router/router.h"
 #include "envoy/ssl/connection.h"
 #include "envoy/stats/scope.h"
+#include "envoy/stream_info/filter_state.h"
 #include "envoy/tracing/http_tracer.h"
 
 #include "common/buffer/buffer_impl.h"
@@ -485,7 +486,8 @@ ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connect
       stream_id_(connection_manager.random_generator_.random()),
       request_response_timespan_(new Stats::HistogramCompletableTimespanImpl(
           connection_manager_.stats_.named_.downstream_rq_time_, connection_manager_.timeSource())),
-      stream_info_(connection_manager_.codec_->protocol(), connection_manager_.timeSource()),
+      stream_info_(connection_manager_.codec_->protocol(), connection_manager_.timeSource(),
+                   connection_manager.filterState()),
       upstream_options_(std::make_shared<Network::Socket::Options>()) {
   ASSERT(!connection_manager.config_.isRoutable() ||
              ((connection_manager.config_.routeConfigProvider() == nullptr &&
@@ -2244,6 +2246,17 @@ bool ConnectionManagerImpl::ActiveStreamDecoderFilter::recreateStream() {
   parent_.connection_manager_.doEndStream(this->parent_);
 
   StreamDecoder& new_stream = parent_.connection_manager_.newStream(*response_encoder, true);
+  // We don't need to copy over the old parent FilterState from the old StreamInfo if it did not
+  // store any objects with a LifeSpan at or above DownstreamRequest. This is to avoid unnecessary
+  // heap allocation.
+  if (parent_.stream_info_.filter_state_->hasDataAtOrAboveLifeSpan(
+          StreamInfo::FilterState::LifeSpan::DownstreamRequest)) {
+    (*parent_.connection_manager_.streams_.begin())->stream_info_.filter_state_ =
+        std::make_shared<StreamInfo::FilterStateImpl>(
+            parent_.stream_info_.filter_state_->parent(),
+            StreamInfo::FilterState::LifeSpan::FilterChain);
+  }
+
   new_stream.decodeHeaders(std::move(request_headers), true);
   return true;
 }
diff --git a/source/common/http/conn_manager_impl.h b/source/common/http/conn_manager_impl.h
index 99ffee1d3b6f0..ec9f58326661b 100644
--- a/source/common/http/conn_manager_impl.h
+++ b/source/common/http/conn_manager_impl.h
@@ -90,6 +90,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
   TimeSource& timeSource() { return time_source_; }
 
+  // Return a reference to the shared_ptr so that it can be lazy created on demand.
+  std::shared_ptr<StreamInfo::FilterState>& filterState() { return filter_state_; }
+
 private:
   struct ActiveStream;
 
@@ -701,6 +704,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
   const Server::OverloadActionState& overload_stop_accepting_requests_ref_;
   const Server::OverloadActionState& overload_disable_keepalive_ref_;
   TimeSource& time_source_;
+  std::shared_ptr<StreamInfo::FilterState> filter_state_;
 };
 
 } // namespace Http
diff --git a/source/common/stream_info/filter_state_impl.cc b/source/common/stream_info/filter_state_impl.cc
index c00f75fe79de6..6097f02e04286 100644
--- a/source/common/stream_info/filter_state_impl.cc
+++ b/source/common/stream_info/filter_state_impl.cc
@@ -5,8 +5,21 @@
 namespace Envoy {
 namespace StreamInfo {
 
-void FilterStateImpl::setData(absl::string_view data_name, std::unique_ptr<Object>&& data,
-                              FilterState::StateType state_type) {
+void FilterStateImpl::setData(absl::string_view data_name, std::shared_ptr<Object> data,
+                              FilterState::StateType state_type, FilterState::LifeSpan life_span) {
+  if (life_span > life_span_) {
+    if (hasDataWithNameInternally(data_name)) {
+      throw EnvoyException(
+          "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
+    }
+    maybeCreateParent(ParentAccessMode::ReadWrite);
+    parent_->setData(data_name, data, state_type, life_span);
+    return;
+  }
+  if (parent_ && parent_->hasDataWithName(data_name)) {
+    throw EnvoyException(
+        "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
+  }
   const auto& it = data_storage_.find(data_name);
   if (it != data_storage_.end()) {
     // We have another object with same data_name. Check for mutability
@@ -23,13 +36,13 @@ void FilterStateImpl::setData(absl::string_view data_name, std::unique_ptr<Objec
   }
 
   std::unique_ptr<FilterStateImpl::FilterObject> filter_object(new FilterStateImpl::FilterObject());
-  filter_object->data_ = std::move(data);
+  filter_object->data_ = data;
   filter_object->state_type_ = state_type;
   data_storage_[data_name] = std::move(filter_object);
 }
 
 bool FilterStateImpl::hasDataWithName(absl::string_view data_name) const {
-  return data_storage_.count(data_name) > 0;
+  return hasDataWithNameInternally(data_name) || (parent_ && parent_->hasDataWithName(data_name));
 }
 
 const FilterState::Object*
@@ -37,6 +50,9 @@ FilterStateImpl::getDataReadOnlyGeneric(absl::string_view data_name) const {
   const auto& it = data_storage_.find(data_name);
 
   if (it == data_storage_.end()) {
+    if (parent_) {
+      return &(parent_->getDataReadOnly<FilterState::Object>(data_name));
+    }
     throw EnvoyException("FilterState::getDataReadOnly<T> called for unknown data name.");
   }
 
@@ -48,6 +64,9 @@ FilterState::Object* FilterStateImpl::getDataMutableGeneric(absl::string_view da
   const auto& it = data_storage_.find(data_name);
 
   if (it == data_storage_.end()) {
+    if (parent_) {
+      return &(parent_->getDataMutable<FilterState::Object>(data_name));
+    }
     throw EnvoyException("FilterState::getDataMutable<T> called for unknown data name.");
   }
 
@@ -60,5 +79,54 @@ FilterState::Object* FilterStateImpl::getDataMutableGeneric(absl::string_view da
   return current->data_.get();
 }
 
+bool FilterStateImpl::hasDataAtOrAboveLifeSpan(FilterState::LifeSpan life_span) const {
+  if (life_span > life_span_) {
+    return parent_ && parent_->hasDataAtOrAboveLifeSpan(life_span);
+  }
+  return !data_storage_.empty() || (parent_ && parent_->hasDataAtOrAboveLifeSpan(life_span));
+}
+
+bool FilterStateImpl::hasDataWithNameInternally(absl::string_view data_name) const {
+  return data_storage_.count(data_name) > 0;
+}
+
+void FilterStateImpl::maybeCreateParent(ParentAccessMode parent_access_mode) {
+  if (parent_ != nullptr) {
+    return;
+  }
+  if (life_span_ >= FilterState::LifeSpan::TopSpan) {
+    return;
+  }
+  if (absl::holds_alternative<std::shared_ptr<FilterState>>(ancestor_)) {
+    std::shared_ptr<FilterState> ancestor = absl::get<std::shared_ptr<FilterState>>(ancestor_);
+    if (ancestor == nullptr || ancestor->lifeSpan() != life_span_ + 1) {
+      parent_ = std::make_shared<FilterStateImpl>(ancestor, FilterState::LifeSpan(life_span_ + 1));
+    } else {
+      parent_ = ancestor;
+    }
+    return;
+  }
+
+  auto lazy_create_ancestor = absl::get<LazyCreateAncestor>(ancestor_);
+  // If we're only going to read data from our parent, we don't need to create lazy ancestor,
+  // because they're empty anyways.
+  if (parent_access_mode == ParentAccessMode::ReadOnly && lazy_create_ancestor.first == nullptr) {
+    return;
+  }
+
+  // Lazy ancestor is not our immediate parent.
+  if (lazy_create_ancestor.second != life_span_ + 1) {
+    parent_ = std::make_shared<FilterStateImpl>(lazy_create_ancestor,
+                                                FilterState::LifeSpan(life_span_ + 1));
+    return;
+  }
+  // Lazy parent is our immediate parent.
+  if (lazy_create_ancestor.first == nullptr) {
+    lazy_create_ancestor.first =
+        std::make_shared<FilterStateImpl>(FilterState::LifeSpan(life_span_ + 1));
+  }
+  parent_ = lazy_create_ancestor.first;
+}
+
 } // namespace StreamInfo
 } // namespace Envoy
diff --git a/source/common/stream_info/filter_state_impl.h b/source/common/stream_info/filter_state_impl.h
index 6f498e2e68ac1..6bf8fb9ad517b 100644
--- a/source/common/stream_info/filter_state_impl.h
+++ b/source/common/stream_info/filter_state_impl.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <memory>
+#include <utility>
 #include <vector>
 
 #include "envoy/stream_info/filter_state.h"
@@ -13,19 +14,57 @@ namespace StreamInfo {
 
 class FilterStateImpl : public FilterState {
 public:
+  FilterStateImpl(FilterState::LifeSpan life_span) : life_span_(life_span) {
+    maybeCreateParent(ParentAccessMode::ReadOnly);
+  }
+
+  /**
+   * @param ancestor a std::shared_ptr storing an already created ancestor.
+   * @param life_span the life span this is handling.
+   */
+  FilterStateImpl(std::shared_ptr<FilterState> ancestor, FilterState::LifeSpan life_span)
+      : ancestor_(ancestor), life_span_(life_span) {
+    maybeCreateParent(ParentAccessMode::ReadOnly);
+  }
+
+  using LazyCreateAncestor = std::pair<std::shared_ptr<FilterState>&, FilterState::LifeSpan>;
+  /**
+   * @param ancestor a std::pair storing an ancestor, that can be passed in as a way to lazy
+   * initialize a FilterState that's owned by an object with bigger scope than this. This is to
+   * avoid creating a FilterState that's empty in most cases.
+   * @param life_span the life span this is handling.
+   */
+  FilterStateImpl(LazyCreateAncestor lazy_create_ancestor, FilterState::LifeSpan life_span)
+      : ancestor_(lazy_create_ancestor), life_span_(life_span) {
+    maybeCreateParent(ParentAccessMode::ReadOnly);
+  }
+
   // FilterState
-  void setData(absl::string_view data_name, std::unique_ptr<Object>&& data,
-               FilterState::StateType state_type) override;
+  void setData(absl::string_view data_name, std::shared_ptr<Object> data,
+               FilterState::StateType state_type,
+               FilterState::LifeSpan life_span = FilterState::LifeSpan::FilterChain) override;
   bool hasDataWithName(absl::string_view) const override;
   const Object* getDataReadOnlyGeneric(absl::string_view data_name) const override;
   Object* getDataMutableGeneric(absl::string_view data_name) override;
+  bool hasDataAtOrAboveLifeSpan(FilterState::LifeSpan life_span) const override;
+
+  FilterState::LifeSpan lifeSpan() const override { return life_span_; }
+  std::shared_ptr<FilterState> parent() const override { return parent_; }
 
 private:
+  // This only checks the local data_storage_ for data_name existence.
+  bool hasDataWithNameInternally(absl::string_view data_name) const;
+  enum class ParentAccessMode { ReadOnly, ReadWrite };
+  void maybeCreateParent(ParentAccessMode parent_access_mode);
+
   struct FilterObject {
-    std::unique_ptr<Object> data_;
+    std::shared_ptr<Object> data_;
     FilterState::StateType state_type_;
   };
 
+  absl::variant<std::shared_ptr<FilterState>, LazyCreateAncestor> ancestor_;
+  std::shared_ptr<FilterState> parent_;
+  const FilterState::LifeSpan life_span_;
   absl::flat_hash_map<std::string, std::unique_ptr<FilterObject>> data_storage_;
 };
 
diff --git a/source/common/stream_info/stream_info_impl.h b/source/common/stream_info/stream_info_impl.h
index 74904bd298620..d07c96bd729c3 100644
--- a/source/common/stream_info/stream_info_impl.h
+++ b/source/common/stream_info/stream_info_impl.h
@@ -15,13 +15,24 @@ namespace Envoy {
 namespace StreamInfo {
 
 struct StreamInfoImpl : public StreamInfo {
-  explicit StreamInfoImpl(TimeSource& time_source)
+  StreamInfoImpl(TimeSource& time_source)
       : time_source_(time_source), start_time_(time_source.systemTime()),
-        start_time_monotonic_(time_source.monotonicTime()) {}
+        start_time_monotonic_(time_source.monotonicTime()),
+        filter_state_(std::make_shared<FilterStateImpl>(FilterState::LifeSpan::FilterChain)) {}
 
-  StreamInfoImpl(Http::Protocol protocol, TimeSource& time_source) : StreamInfoImpl(time_source) {
-    protocol_ = protocol;
-  }
+  StreamInfoImpl(Http::Protocol protocol, TimeSource& time_source)
+      : time_source_(time_source), start_time_(time_source.systemTime()),
+        start_time_monotonic_(time_source.monotonicTime()), protocol_(protocol),
+        filter_state_(std::make_shared<FilterStateImpl>(FilterState::LifeSpan::FilterChain)) {}
+
+  StreamInfoImpl(Http::Protocol protocol, TimeSource& time_source,
+                 std::shared_ptr<FilterState>& parent_filter_state)
+      : time_source_(time_source), start_time_(time_source.systemTime()),
+        start_time_monotonic_(time_source.monotonicTime()), protocol_(protocol),
+        filter_state_(std::make_shared<FilterStateImpl>(
+            FilterStateImpl::LazyCreateAncestor(parent_filter_state,
+                                                FilterState::LifeSpan::DownstreamConnection),
+            FilterState::LifeSpan::FilterChain)) {}
 
   SystemTime startTime() const override { return start_time_; }
 
@@ -204,8 +215,8 @@ struct StreamInfoImpl : public StreamInfo {
     (*metadata_.mutable_filter_metadata())[name].MergeFrom(value);
   };
 
-  FilterState& filterState() override { return filter_state_; }
-  const FilterState& filterState() const override { return filter_state_; }
+  FilterState& filterState() override { return *filter_state_; }
+  const FilterState& filterState() const override { return *filter_state_; }
 
   void setRequestedServerName(absl::string_view requested_server_name) override {
     requested_server_name_ = std::string(requested_server_name);
@@ -249,7 +260,7 @@ struct StreamInfoImpl : public StreamInfo {
   bool health_check_request_{};
   const Router::RouteEntry* route_entry_{};
   envoy::api::v2::core::Metadata metadata_{};
-  FilterStateImpl filter_state_{};
+  std::shared_ptr<FilterStateImpl> filter_state_;
   std::string route_name_;
 
 private:
diff --git a/source/extensions/filters/http/grpc_stats/grpc_stats_filter.cc b/source/extensions/filters/http/grpc_stats/grpc_stats_filter.cc
index 033a649fb6b80..a15cfae09f026 100644
--- a/source/extensions/filters/http/grpc_stats/grpc_stats_filter.cc
+++ b/source/extensions/filters/http/grpc_stats/grpc_stats_filter.cc
@@ -81,7 +81,8 @@ class GrpcStatsFilter : public Http::PassThroughFilter {
       filter_object_ = state.get();
       decoder_callbacks_->streamInfo().filterState().setData(
           HttpFilterNames::get().GrpcStats, std::move(state),
-          StreamInfo::FilterState::StateType::Mutable);
+          StreamInfo::FilterState::StateType::Mutable,
+          StreamInfo::FilterState::LifeSpan::FilterChain);
     }
     filter_object_->request_message_count = request_counter_.frameCount();
     filter_object_->response_message_count = response_counter_.frameCount();
diff --git a/source/extensions/filters/network/sni_cluster/sni_cluster.cc b/source/extensions/filters/network/sni_cluster/sni_cluster.cc
index 2b403b586a198..b0962fb198816 100644
--- a/source/extensions/filters/network/sni_cluster/sni_cluster.cc
+++ b/source/extensions/filters/network/sni_cluster/sni_cluster.cc
@@ -21,7 +21,8 @@ Network::FilterStatus SniClusterFilter::onNewConnection() {
     read_callbacks_->connection().streamInfo().filterState().setData(
         TcpProxy::PerConnectionCluster::key(),
         std::make_unique<TcpProxy::PerConnectionCluster>(sni),
-        StreamInfo::FilterState::StateType::Mutable);
+        StreamInfo::FilterState::StateType::Mutable,
+        StreamInfo::FilterState::LifeSpan::DownstreamConnection);
   }
 
   return Network::FilterStatus::Continue;
diff --git a/test/common/access_log/access_log_formatter_test.cc b/test/common/access_log/access_log_formatter_test.cc
index dbcfbcde89ba8..9fabf6295175b 100644
--- a/test/common/access_log/access_log_formatter_test.cc
+++ b/test/common/access_log/access_log_formatter_test.cc
@@ -1472,10 +1472,12 @@ TEST(AccessLogFormatterTest, CompositeFormatterSuccess) {
     EXPECT_CALL(Const(stream_info), filterState()).Times(testing::AtLeast(1));
     stream_info.filter_state_.setData("testing",
                                       std::make_unique<Router::StringAccessorImpl>("test_value"),
-                                      StreamInfo::FilterState::StateType::ReadOnly);
+                                      StreamInfo::FilterState::StateType::ReadOnly,
+                                      StreamInfo::FilterState::LifeSpan::FilterChain);
     stream_info.filter_state_.setData("serialized",
                                       std::make_unique<TestSerializedUnknownFilterState>(),
-                                      StreamInfo::FilterState::StateType::ReadOnly);
+                                      StreamInfo::FilterState::StateType::ReadOnly,
+                                      StreamInfo::FilterState::LifeSpan::FilterChain);
     const std::string format = "%FILTER_STATE(testing)%|%FILTER_STATE(serialized)%|"
                                "%FILTER_STATE(testing):8%|%FILTER_STATE(nonexisting)%";
     FormatterImpl formatter(format);
diff --git a/test/common/http/conn_manager_impl_test.cc b/test/common/http/conn_manager_impl_test.cc
index 3dcc014b9f799..86239175431e9 100644
--- a/test/common/http/conn_manager_impl_test.cc
+++ b/test/common/http/conn_manager_impl_test.cc
@@ -136,35 +136,40 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan
     }
   }
 
-  void setupFilterChain(int num_decoder_filters, int num_encoder_filters) {
+  void setupFilterChain(int num_decoder_filters, int num_encoder_filters, int num_requests = 1) {
     // NOTE: The length/repetition in this routine allows InSequence to work correctly in an outer
     // scope.
-    for (int i = 0; i < num_decoder_filters; i++) {
+    for (int i = 0; i < num_decoder_filters * num_requests; i++) {
       decoder_filters_.push_back(new MockStreamDecoderFilter());
     }
 
-    for (int i = 0; i < num_encoder_filters; i++) {
+    for (int i = 0; i < num_encoder_filters * num_requests; i++) {
       encoder_filters_.push_back(new MockStreamEncoderFilter());
     }
 
-    EXPECT_CALL(filter_factory_, createFilterChain(_))
-        .WillOnce(Invoke([num_decoder_filters, num_encoder_filters,
-                          this](FilterChainFactoryCallbacks& callbacks) -> void {
-          for (int i = 0; i < num_decoder_filters; i++) {
-            callbacks.addStreamDecoderFilter(StreamDecoderFilterSharedPtr{decoder_filters_[i]});
-          }
-
-          for (int i = 0; i < num_encoder_filters; i++) {
-            callbacks.addStreamEncoderFilter(StreamEncoderFilterSharedPtr{encoder_filters_[i]});
-          }
-        }));
+    InSequence s;
+    for (int req = 0; req < num_requests; req++) {
+      EXPECT_CALL(filter_factory_, createFilterChain(_))
+          .WillOnce(Invoke([num_decoder_filters, num_encoder_filters, req,
+                            this](FilterChainFactoryCallbacks& callbacks) -> void {
+            for (int i = 0; i < num_decoder_filters; i++) {
+              callbacks.addStreamDecoderFilter(
+                  StreamDecoderFilterSharedPtr{decoder_filters_[req * num_decoder_filters + i]});
+            }
+
+            for (int i = 0; i < num_encoder_filters; i++) {
+              callbacks.addStreamEncoderFilter(
+                  StreamEncoderFilterSharedPtr{encoder_filters_[req * num_encoder_filters + i]});
+            }
+          }));
 
-    for (int i = 0; i < num_decoder_filters; i++) {
-      EXPECT_CALL(*decoder_filters_[i], setDecoderFilterCallbacks(_));
-    }
+      for (int i = 0; i < num_decoder_filters; i++) {
+        EXPECT_CALL(*decoder_filters_[req * num_decoder_filters + i], setDecoderFilterCallbacks(_));
+      }
 
-    for (int i = 0; i < num_encoder_filters; i++) {
-      EXPECT_CALL(*encoder_filters_[i], setEncoderFilterCallbacks(_));
+      for (int i = 0; i < num_encoder_filters; i++) {
+        EXPECT_CALL(*encoder_filters_[req * num_encoder_filters + i], setEncoderFilterCallbacks(_));
+      }
     }
   }
 
@@ -5060,6 +5065,87 @@ TEST_F(HttpConnectionManagerImplTest, HeaderOnlyRequestAndResponseUsingHttp3) {
   EXPECT_EQ(0U, stats_.named_.downstream_cx_http3_active_.value());
 }
 
+namespace {
+
+class SimpleType : public StreamInfo::FilterState::Object {
+public:
+  SimpleType(int value) : value_(value) {}
+  int access() const { return value_; }
+
+private:
+  int value_;
+};
+
+} // namespace
+
+TEST_F(HttpConnectionManagerImplTest, ConnectionFilterState) {
+  setup(false, "envoy-custom-server", false);
+
+  setupFilterChain(1, 0, /* num_requests = */ 3);
+
+  EXPECT_CALL(*codec_, dispatch(_)).Times(2).WillRepeatedly(Invoke([&](Buffer::Instance&) -> void {
+    StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
+    HeaderMapPtr headers{
+        new TestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
+    decoder->decodeHeaders(std::move(headers), true);
+  }));
+  {
+    InSequence s;
+    EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, true))
+        .WillOnce(Invoke([this](HeaderMap&, bool) -> FilterHeadersStatus {
+          decoder_filters_[0]->callbacks_->streamInfo().filterState().setData(
+              "per_filter_chain", std::make_unique<SimpleType>(1),
+              StreamInfo::FilterState::StateType::ReadOnly,
+              StreamInfo::FilterState::LifeSpan::FilterChain);
+          decoder_filters_[0]->callbacks_->streamInfo().filterState().setData(
+              "per_downstream_request", std::make_unique<SimpleType>(2),
+              StreamInfo::FilterState::StateType::ReadOnly,
+              StreamInfo::FilterState::LifeSpan::DownstreamRequest);
+          decoder_filters_[0]->callbacks_->streamInfo().filterState().setData(
+              "per_downstream_connection", std::make_unique<SimpleType>(3),
+              StreamInfo::FilterState::StateType::ReadOnly,
+              StreamInfo::FilterState::LifeSpan::DownstreamConnection);
+          return FilterHeadersStatus::StopIteration;
+        }));
+    EXPECT_CALL(*decoder_filters_[1], decodeHeaders(_, true))
+        .WillOnce(Invoke([this](HeaderMap&, bool) -> FilterHeadersStatus {
+          EXPECT_FALSE(
+              decoder_filters_[1]->callbacks_->streamInfo().filterState().hasData<SimpleType>(
+                  "per_filter_chain"));
+          EXPECT_TRUE(
+              decoder_filters_[1]->callbacks_->streamInfo().filterState().hasData<SimpleType>(
+                  "per_downstream_request"));
+          EXPECT_TRUE(
+              decoder_filters_[1]->callbacks_->streamInfo().filterState().hasData<SimpleType>(
+                  "per_downstream_connection"));
+          return FilterHeadersStatus::StopIteration;
+        }));
+    EXPECT_CALL(*decoder_filters_[2], decodeHeaders(_, true))
+        .WillOnce(Invoke([this](HeaderMap&, bool) -> FilterHeadersStatus {
+          EXPECT_FALSE(
+              decoder_filters_[2]->callbacks_->streamInfo().filterState().hasData<SimpleType>(
+                  "per_filter_chain"));
+          EXPECT_FALSE(
+              decoder_filters_[2]->callbacks_->streamInfo().filterState().hasData<SimpleType>(
+                  "per_downstream_request"));
+          EXPECT_TRUE(
+              decoder_filters_[2]->callbacks_->streamInfo().filterState().hasData<SimpleType>(
+                  "per_downstream_connection"));
+          return FilterHeadersStatus::StopIteration;
+        }));
+  }
+
+  EXPECT_CALL(*decoder_filters_[0], decodeComplete());
+  EXPECT_CALL(*decoder_filters_[0], onDestroy());
+  EXPECT_CALL(*decoder_filters_[1], decodeComplete());
+  EXPECT_CALL(*decoder_filters_[2], decodeComplete());
+
+  Buffer::OwnedImpl fake_input;
+  conn_manager_->onData(fake_input, false);
+  decoder_filters_[0]->callbacks_->recreateStream();
+  conn_manager_->onData(fake_input, false);
+}
+
 class HttpConnectionManagerImplDeathTest : public HttpConnectionManagerImplTest {
 public:
   Router::RouteConfigProvider* routeConfigProvider() override {
diff --git a/test/common/network/transport_socket_options_impl_test.cc b/test/common/network/transport_socket_options_impl_test.cc
index e5b2eaa08d2c2..d330a8edf8d3c 100644
--- a/test/common/network/transport_socket_options_impl_test.cc
+++ b/test/common/network/transport_socket_options_impl_test.cc
@@ -10,22 +10,26 @@ namespace Network {
 namespace {
 
 class TransportSocketOptionsImplTest : public testing::Test {
+public:
+  TransportSocketOptionsImplTest()
+      : filter_state_(StreamInfo::FilterState::LifeSpan::FilterChain) {}
+
 protected:
   StreamInfo::FilterStateImpl filter_state_;
 };
 
 TEST_F(TransportSocketOptionsImplTest, Nullptr) {
   EXPECT_EQ(nullptr, TransportSocketOptionsUtility::fromFilterState(filter_state_));
-  filter_state_.setData("random_key_has_no_effect",
-                        std::make_unique<UpstreamServerName>("www.example.com"),
-                        StreamInfo::FilterState::StateType::ReadOnly);
+  filter_state_.setData(
+      "random_key_has_no_effect", std::make_unique<UpstreamServerName>("www.example.com"),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::FilterChain);
   EXPECT_EQ(nullptr, TransportSocketOptionsUtility::fromFilterState(filter_state_));
 }
 
 TEST_F(TransportSocketOptionsImplTest, UpstreamServer) {
-  filter_state_.setData(UpstreamServerName::key(),
-                        std::make_unique<UpstreamServerName>("www.example.com"),
-                        StreamInfo::FilterState::StateType::ReadOnly);
+  filter_state_.setData(
+      UpstreamServerName::key(), std::make_unique<UpstreamServerName>("www.example.com"),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::FilterChain);
   auto transport_socket_options = TransportSocketOptionsUtility::fromFilterState(filter_state_);
   EXPECT_EQ(absl::make_optional<std::string>("www.example.com"),
             transport_socket_options->serverNameOverride());
@@ -34,9 +38,9 @@ TEST_F(TransportSocketOptionsImplTest, UpstreamServer) {
 
 TEST_F(TransportSocketOptionsImplTest, ApplicationProtocols) {
   std::vector<std::string> http_alpns{"h2", "http/1.1"};
-  filter_state_.setData(ApplicationProtocols::key(),
-                        std::make_unique<ApplicationProtocols>(http_alpns),
-                        StreamInfo::FilterState::StateType::ReadOnly);
+  filter_state_.setData(
+      ApplicationProtocols::key(), std::make_unique<ApplicationProtocols>(http_alpns),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::FilterChain);
   auto transport_socket_options = TransportSocketOptionsUtility::fromFilterState(filter_state_);
   EXPECT_EQ(absl::nullopt, transport_socket_options->serverNameOverride());
   EXPECT_EQ(http_alpns, transport_socket_options->applicationProtocolListOverride());
@@ -44,12 +48,12 @@ TEST_F(TransportSocketOptionsImplTest, ApplicationProtocols) {
 
 TEST_F(TransportSocketOptionsImplTest, Both) {
   std::vector<std::string> http_alpns{"h2", "http/1.1"};
-  filter_state_.setData(UpstreamServerName::key(),
-                        std::make_unique<UpstreamServerName>("www.example.com"),
-                        StreamInfo::FilterState::StateType::ReadOnly);
-  filter_state_.setData(ApplicationProtocols::key(),
-                        std::make_unique<ApplicationProtocols>(http_alpns),
-                        StreamInfo::FilterState::StateType::ReadOnly);
+  filter_state_.setData(
+      UpstreamServerName::key(), std::make_unique<UpstreamServerName>("www.example.com"),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::FilterChain);
+  filter_state_.setData(
+      ApplicationProtocols::key(), std::make_unique<ApplicationProtocols>(http_alpns),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::FilterChain);
   auto transport_socket_options = TransportSocketOptionsUtility::fromFilterState(filter_state_);
   EXPECT_EQ(absl::make_optional<std::string>("www.example.com"),
             transport_socket_options->serverNameOverride());
diff --git a/test/common/router/header_formatter_test.cc b/test/common/router/header_formatter_test.cc
index 2e5c9aacf181b..6c210a47b145c 100644
--- a/test/common/router/header_formatter_test.cc
+++ b/test/common/router/header_formatter_test.cc
@@ -542,9 +542,11 @@ TEST_F(StreamInfoHeaderFormatterTest, TestFormatWithUpstreamMetadataVariableMiss
 }
 
 TEST_F(StreamInfoHeaderFormatterTest, TestFormatWithPerRequestStateVariable) {
-  Envoy::StreamInfo::FilterStateImpl filter_state;
+  Envoy::StreamInfo::FilterStateImpl filter_state(
+      Envoy::StreamInfo::FilterState::LifeSpan::FilterChain);
   filter_state.setData("testing", std::make_unique<StringAccessorImpl>("test_value"),
-                       StreamInfo::FilterState::StateType::ReadOnly);
+                       StreamInfo::FilterState::StateType::ReadOnly,
+                       StreamInfo::FilterState::LifeSpan::FilterChain);
   EXPECT_EQ("test_value", filter_state.getDataReadOnly<StringAccessor>("testing").asString());
 
   NiceMock<Envoy::StreamInfo::MockStreamInfo> stream_info;
@@ -557,9 +559,11 @@ TEST_F(StreamInfoHeaderFormatterTest, TestFormatWithPerRequestStateVariable) {
 }
 
 TEST_F(StreamInfoHeaderFormatterTest, TestFormatWithNonStringPerRequestStateVariable) {
-  Envoy::StreamInfo::FilterStateImpl filter_state;
+  Envoy::StreamInfo::FilterStateImpl filter_state(
+      Envoy::StreamInfo::FilterState::LifeSpan::FilterChain);
   filter_state.setData("testing", std::make_unique<StreamInfo::TestIntAccessor>(1),
-                       StreamInfo::FilterState::StateType::ReadOnly);
+                       StreamInfo::FilterState::StateType::ReadOnly,
+                       StreamInfo::FilterState::LifeSpan::FilterChain);
   EXPECT_EQ(1, filter_state.getDataReadOnly<StreamInfo::TestIntAccessor>("testing").access());
 
   NiceMock<Envoy::StreamInfo::MockStreamInfo> stream_info;
@@ -811,9 +815,11 @@ TEST(HeaderParserTest, TestParseInternal) {
   const SystemTime start_time(std::chrono::milliseconds(1522796769123));
   ON_CALL(stream_info, startTime()).WillByDefault(Return(start_time));
 
-  Envoy::StreamInfo::FilterStateImpl filter_state;
+  Envoy::StreamInfo::FilterStateImpl filter_state(
+      Envoy::StreamInfo::FilterState::LifeSpan::FilterChain);
   filter_state.setData("testing", std::make_unique<StringAccessorImpl>("test_value"),
-                       StreamInfo::FilterState::StateType::ReadOnly);
+                       StreamInfo::FilterState::StateType::ReadOnly,
+                       StreamInfo::FilterState::LifeSpan::FilterChain);
   ON_CALL(stream_info, filterState()).WillByDefault(ReturnRef(filter_state));
   ON_CALL(Const(stream_info), filterState()).WillByDefault(ReturnRef(filter_state));
 
@@ -979,9 +985,11 @@ request_headers_to_remove: ["x-nope"]
       )EOF"));
   ON_CALL(*host, metadata()).WillByDefault(Return(metadata));
 
-  Envoy::StreamInfo::FilterStateImpl filter_state;
+  Envoy::StreamInfo::FilterStateImpl filter_state(
+      Envoy::StreamInfo::FilterState::LifeSpan::FilterChain);
   filter_state.setData("testing", std::make_unique<StringAccessorImpl>("test_value"),
-                       StreamInfo::FilterState::StateType::ReadOnly);
+                       StreamInfo::FilterState::StateType::ReadOnly,
+                       StreamInfo::FilterState::LifeSpan::FilterChain);
   ON_CALL(stream_info, filterState()).WillByDefault(ReturnRef(filter_state));
   ON_CALL(Const(stream_info), filterState()).WillByDefault(ReturnRef(filter_state));
 
diff --git a/test/common/router/router_test.cc b/test/common/router/router_test.cc
index e1e5856308446..5dcddfc3bebbf 100644
--- a/test/common/router/router_test.cc
+++ b/test/common/router/router_test.cc
@@ -795,7 +795,8 @@ void RouterTestBase::testAppendCluster(absl::optional<Http::LowerCaseString> clu
       /* do_not_forward */ false,
       /* not_forwarded_header */ absl::nullopt);
   callbacks_.streamInfo().filterState().setData(DebugConfig::key(), std::move(debug_config),
-                                                StreamInfo::FilterState::StateType::ReadOnly);
+                                                StreamInfo::FilterState::StateType::ReadOnly,
+                                                StreamInfo::FilterState::LifeSpan::FilterChain);
 
   NiceMock<Http::MockStreamEncoder> encoder;
   Http::StreamDecoder* response_decoder = nullptr;
@@ -846,7 +847,8 @@ void RouterTestBase::testAppendUpstreamHost(
       /* do_not_forward */ false,
       /* not_forwarded_header */ absl::nullopt);
   callbacks_.streamInfo().filterState().setData(DebugConfig::key(), std::move(debug_config),
-                                                StreamInfo::FilterState::StateType::ReadOnly);
+                                                StreamInfo::FilterState::StateType::ReadOnly,
+                                                StreamInfo::FilterState::LifeSpan::FilterChain);
   cm_.conn_pool_.host_->hostname_ = "scooby.doo";
 
   NiceMock<Http::MockStreamEncoder> encoder;
@@ -917,7 +919,8 @@ void RouterTestBase::testDoNotForward(
       /* do_not_forward */ true,
       /* not_forwarded_header */ not_forwarded_header_name);
   callbacks_.streamInfo().filterState().setData(DebugConfig::key(), std::move(debug_config),
-                                                StreamInfo::FilterState::StateType::ReadOnly);
+                                                StreamInfo::FilterState::StateType::ReadOnly,
+                                                StreamInfo::FilterState::LifeSpan::FilterChain);
 
   Http::TestHeaderMapImpl response_headers{
       {":status", "204"},
@@ -949,7 +952,8 @@ TEST_F(RouterTest, AllDebugConfig) {
       /* do_not_forward */ true,
       /* not_forwarded_header */ absl::nullopt);
   callbacks_.streamInfo().filterState().setData(DebugConfig::key(), std::move(debug_config),
-                                                StreamInfo::FilterState::StateType::ReadOnly);
+                                                StreamInfo::FilterState::StateType::ReadOnly,
+                                                StreamInfo::FilterState::LifeSpan::FilterChain);
   cm_.conn_pool_.host_->hostname_ = "scooby.doo";
 
   Http::TestHeaderMapImpl response_headers{{":status", "204"},
@@ -4327,7 +4331,7 @@ TEST_F(RouterTest, ApplicationProtocols) {
   callbacks_.streamInfo().filterState().setData(
       Network::ApplicationProtocols::key(),
       std::make_unique<Network::ApplicationProtocols>(std::vector<std::string>{"foo", "bar"}),
-      StreamInfo::FilterState::StateType::ReadOnly);
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::FilterChain);
 
   EXPECT_CALL(cm_, httpConnPoolForCluster(_, _, _, _))
       .WillOnce(
diff --git a/test/common/stream_info/filter_state_impl_test.cc b/test/common/stream_info/filter_state_impl_test.cc
index 4cf7ab2508814..4c2659463adae 100644
--- a/test/common/stream_info/filter_state_impl_test.cc
+++ b/test/common/stream_info/filter_state_impl_test.cc
@@ -48,7 +48,9 @@ class FilterStateImplTest : public testing::Test {
 public:
   FilterStateImplTest() { resetFilterState(); }
 
-  void resetFilterState() { filter_state_ = std::make_unique<FilterStateImpl>(); }
+  void resetFilterState() {
+    filter_state_ = std::make_unique<FilterStateImpl>(FilterState::LifeSpan::FilterChain);
+  }
   FilterState& filter_state() { return *filter_state_; }
 
 private:
@@ -62,7 +64,7 @@ TEST_F(FilterStateImplTest, Simple) {
   size_t destruction_count = 0u;
   filter_state().setData(
       "test_name", std::make_unique<TestStoredTypeTracking>(5, &access_count, &destruction_count),
-      FilterState::StateType::ReadOnly);
+      FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   EXPECT_EQ(0u, access_count);
   EXPECT_EQ(0u, destruction_count);
 
@@ -85,11 +87,11 @@ TEST_F(FilterStateImplTest, SameTypes) {
   filter_state().setData(
       "test_1",
       std::make_unique<TestStoredTypeTracking>(ValueOne, &access_count_1, &destruction_count),
-      FilterState::StateType::ReadOnly);
+      FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   filter_state().setData(
       "test_2",
       std::make_unique<TestStoredTypeTracking>(ValueTwo, &access_count_2, &destruction_count),
-      FilterState::StateType::ReadOnly);
+      FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   EXPECT_EQ(0u, access_count_1);
   EXPECT_EQ(0u, access_count_2);
   EXPECT_EQ(0u, destruction_count);
@@ -106,19 +108,19 @@ TEST_F(FilterStateImplTest, SameTypes) {
 
 TEST_F(FilterStateImplTest, SimpleTypeReadOnly) {
   filter_state().setData("test_1", std::make_unique<SimpleType>(1),
-                         FilterState::StateType::ReadOnly);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   filter_state().setData("test_2", std::make_unique<SimpleType>(2),
-                         FilterState::StateType::ReadOnly);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
 
   EXPECT_EQ(1, filter_state().getDataReadOnly<SimpleType>("test_1").access());
   EXPECT_EQ(2, filter_state().getDataReadOnly<SimpleType>("test_2").access());
 }
 
 TEST_F(FilterStateImplTest, SimpleTypeMutable) {
-  filter_state().setData("test_1", std::make_unique<SimpleType>(1),
-                         FilterState::StateType::Mutable);
-  filter_state().setData("test_2", std::make_unique<SimpleType>(2),
-                         FilterState::StateType::Mutable);
+  filter_state().setData("test_1", std::make_unique<SimpleType>(1), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
+  filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
 
   EXPECT_EQ(1, filter_state().getDataReadOnly<SimpleType>("test_1").access());
   EXPECT_EQ(2, filter_state().getDataReadOnly<SimpleType>("test_2").access());
@@ -132,53 +134,53 @@ TEST_F(FilterStateImplTest, SimpleTypeMutable) {
 TEST_F(FilterStateImplTest, NameConflictReadOnly) {
   // read only data cannot be overwritten (by any state type)
   filter_state().setData("test_1", std::make_unique<SimpleType>(1),
-                         FilterState::StateType::ReadOnly);
-  EXPECT_THROW_WITH_MESSAGE(filter_state().setData("test_1", std::make_unique<SimpleType>(2),
-                                                   FilterState::StateType::ReadOnly),
-                            EnvoyException,
-                            "FilterState::setData<T> called twice on same ReadOnly state.");
-  EXPECT_THROW_WITH_MESSAGE(filter_state().setData("test_1", std::make_unique<SimpleType>(2),
-                                                   FilterState::StateType::Mutable),
-                            EnvoyException,
-                            "FilterState::setData<T> called twice on same ReadOnly state.");
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
+  EXPECT_THROW_WITH_MESSAGE(
+      filter_state().setData("test_1", std::make_unique<SimpleType>(2),
+                             FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain),
+      EnvoyException, "FilterState::setData<T> called twice on same ReadOnly state.");
+  EXPECT_THROW_WITH_MESSAGE(
+      filter_state().setData("test_1", std::make_unique<SimpleType>(2),
+                             FilterState::StateType::Mutable, FilterState::LifeSpan::FilterChain),
+      EnvoyException, "FilterState::setData<T> called twice on same ReadOnly state.");
   EXPECT_EQ(1, filter_state().getDataReadOnly<SimpleType>("test_1").access());
 }
 
 TEST_F(FilterStateImplTest, NameConflictDifferentTypesReadOnly) {
   filter_state().setData("test_1", std::make_unique<SimpleType>(1),
-                         FilterState::StateType::ReadOnly);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   EXPECT_THROW_WITH_MESSAGE(
       filter_state().setData("test_1",
                              std::make_unique<TestStoredTypeTracking>(2, nullptr, nullptr),
-                             FilterState::StateType::ReadOnly),
+                             FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain),
       EnvoyException, "FilterState::setData<T> called twice on same ReadOnly state.");
 }
 
 TEST_F(FilterStateImplTest, NameConflictMutableAndReadOnly) {
   // Mutable data cannot be overwritten by read only data.
-  filter_state().setData("test_1", std::make_unique<SimpleType>(1),
-                         FilterState::StateType::Mutable);
-  EXPECT_THROW_WITH_MESSAGE(filter_state().setData("test_1", std::make_unique<SimpleType>(2),
-                                                   FilterState::StateType::ReadOnly),
-                            EnvoyException,
-                            "FilterState::setData<T> called twice with different state types.");
+  filter_state().setData("test_1", std::make_unique<SimpleType>(1), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
+  EXPECT_THROW_WITH_MESSAGE(
+      filter_state().setData("test_1", std::make_unique<SimpleType>(2),
+                             FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain),
+      EnvoyException, "FilterState::setData<T> called twice with different state types.");
 }
 
 TEST_F(FilterStateImplTest, NoNameConflictMutableAndMutable) {
   // Mutable data can be overwritten by another mutable data of same or different type.
 
   // mutable + mutable - same type
-  filter_state().setData("test_2", std::make_unique<SimpleType>(3),
-                         FilterState::StateType::Mutable);
-  filter_state().setData("test_2", std::make_unique<SimpleType>(4),
-                         FilterState::StateType::Mutable);
+  filter_state().setData("test_2", std::make_unique<SimpleType>(3), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
+  filter_state().setData("test_2", std::make_unique<SimpleType>(4), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
   EXPECT_EQ(4, filter_state().getDataMutable<SimpleType>("test_2").access());
 
   // mutable + mutable - different types
-  filter_state().setData("test_4", std::make_unique<SimpleType>(7),
-                         FilterState::StateType::Mutable);
+  filter_state().setData("test_4", std::make_unique<SimpleType>(7), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
   filter_state().setData("test_4", std::make_unique<TestStoredTypeTracking>(8, nullptr, nullptr),
-                         FilterState::StateType::Mutable);
+                         FilterState::StateType::Mutable, FilterState::LifeSpan::FilterChain);
   EXPECT_EQ(8, filter_state().getDataReadOnly<TestStoredTypeTracking>("test_4").access());
 }
 
@@ -191,7 +193,7 @@ TEST_F(FilterStateImplTest, UnknownName) {
 
 TEST_F(FilterStateImplTest, WrongTypeGet) {
   filter_state().setData("test_name", std::make_unique<TestStoredTypeTracking>(5, nullptr, nullptr),
-                         FilterState::StateType::ReadOnly);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   EXPECT_EQ(5, filter_state().getDataReadOnly<TestStoredTypeTracking>("test_name").access());
   EXPECT_THROW_WITH_MESSAGE(filter_state().getDataReadOnly<SimpleType>("test_name"), EnvoyException,
                             "Data stored under test_name cannot be coerced to specified type");
@@ -200,7 +202,7 @@ TEST_F(FilterStateImplTest, WrongTypeGet) {
 TEST_F(FilterStateImplTest, ErrorAccessingReadOnlyAsMutable) {
   // Accessing read only data as mutable should throw error
   filter_state().setData("test_name", std::make_unique<TestStoredTypeTracking>(5, nullptr, nullptr),
-                         FilterState::StateType::ReadOnly);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   EXPECT_THROW_WITH_MESSAGE(
       filter_state().getDataMutable<TestStoredTypeTracking>("test_name"), EnvoyException,
       "FilterState::getDataMutable<T> tried to access immutable data as mutable.");
@@ -217,12 +219,14 @@ class C : public B {};
 } // namespace
 
 TEST_F(FilterStateImplTest, FungibleInheritance) {
-  filter_state().setData("testB", std::make_unique<B>(), FilterState::StateType::ReadOnly);
+  filter_state().setData("testB", std::make_unique<B>(), FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_state().hasData<B>("testB"));
   EXPECT_TRUE(filter_state().hasData<A>("testB"));
   EXPECT_FALSE(filter_state().hasData<C>("testB"));
 
-  filter_state().setData("testC", std::make_unique<C>(), FilterState::StateType::ReadOnly);
+  filter_state().setData("testC", std::make_unique<C>(), FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_state().hasData<B>("testC"));
   EXPECT_TRUE(filter_state().hasData<A>("testC"));
   EXPECT_TRUE(filter_state().hasData<C>("testC"));
@@ -230,7 +234,7 @@ TEST_F(FilterStateImplTest, FungibleInheritance) {
 
 TEST_F(FilterStateImplTest, HasData) {
   filter_state().setData("test_1", std::make_unique<SimpleType>(1),
-                         FilterState::StateType::ReadOnly);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_state().hasData<SimpleType>("test_1"));
   EXPECT_FALSE(filter_state().hasData<SimpleType>("test_2"));
   EXPECT_FALSE(filter_state().hasData<TestStoredTypeTracking>("test_1"));
@@ -239,5 +243,160 @@ TEST_F(FilterStateImplTest, HasData) {
   EXPECT_FALSE(filter_state().hasDataWithName("test_2"));
 }
 
+TEST_F(FilterStateImplTest, LifeSpanInitFromParent) {
+  filter_state().setData("test_1", std::make_unique<SimpleType>(1),
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
+  filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
+  filter_state().setData("test_3", std::make_unique<SimpleType>(3),
+                         FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::DownstreamRequest);
+  filter_state().setData("test_4", std::make_unique<SimpleType>(4), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamRequest);
+  filter_state().setData("test_5", std::make_unique<SimpleType>(5),
+                         FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::DownstreamConnection);
+  filter_state().setData("test_6", std::make_unique<SimpleType>(6), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamConnection);
+
+  FilterStateImpl new_filter_state(filter_state().parent(), FilterState::LifeSpan::FilterChain);
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_1"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_2"));
+  EXPECT_TRUE(new_filter_state.hasDataWithName("test_3"));
+  EXPECT_TRUE(new_filter_state.hasDataWithName("test_4"));
+  EXPECT_TRUE(new_filter_state.hasDataWithName("test_5"));
+  EXPECT_TRUE(new_filter_state.hasDataWithName("test_6"));
+  EXPECT_THROW_WITH_MESSAGE(
+      new_filter_state.getDataMutable<SimpleType>("test_3"), EnvoyException,
+      "FilterState::getDataMutable<T> tried to access immutable data as mutable.");
+  EXPECT_EQ(4, new_filter_state.getDataMutable<SimpleType>("test_4").access());
+  EXPECT_THROW_WITH_MESSAGE(
+      new_filter_state.getDataMutable<SimpleType>("test_5"), EnvoyException,
+      "FilterState::getDataMutable<T> tried to access immutable data as mutable.");
+  EXPECT_EQ(6, new_filter_state.getDataMutable<SimpleType>("test_6").access());
+}
+
+TEST_F(FilterStateImplTest, LifeSpanInitFromGrandparent) {
+  filter_state().setData("test_1", std::make_unique<SimpleType>(1),
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
+  filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
+  filter_state().setData("test_3", std::make_unique<SimpleType>(3),
+                         FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::DownstreamRequest);
+  filter_state().setData("test_4", std::make_unique<SimpleType>(4), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamRequest);
+  filter_state().setData("test_5", std::make_unique<SimpleType>(5),
+                         FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::DownstreamConnection);
+  filter_state().setData("test_6", std::make_unique<SimpleType>(6), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamConnection);
+
+  FilterStateImpl new_filter_state(filter_state().parent()->parent(),
+                                   FilterState::LifeSpan::FilterChain);
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_1"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_2"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_3"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_4"));
+  EXPECT_TRUE(new_filter_state.hasDataWithName("test_5"));
+  EXPECT_TRUE(new_filter_state.hasDataWithName("test_6"));
+  EXPECT_THROW_WITH_MESSAGE(
+      new_filter_state.getDataMutable<SimpleType>("test_5"), EnvoyException,
+      "FilterState::getDataMutable<T> tried to access immutable data as mutable.");
+  EXPECT_EQ(6, new_filter_state.getDataMutable<SimpleType>("test_6").access());
+}
+
+TEST_F(FilterStateImplTest, LifeSpanInitFromNonParent) {
+  filter_state().setData("test_1", std::make_unique<SimpleType>(1),
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
+  filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::FilterChain);
+  filter_state().setData("test_3", std::make_unique<SimpleType>(3),
+                         FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::DownstreamRequest);
+  filter_state().setData("test_4", std::make_unique<SimpleType>(4), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamRequest);
+  filter_state().setData("test_5", std::make_unique<SimpleType>(5),
+                         FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::DownstreamConnection);
+  filter_state().setData("test_6", std::make_unique<SimpleType>(6), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamConnection);
+
+  FilterStateImpl new_filter_state(filter_state().parent(),
+                                   FilterState::LifeSpan::DownstreamRequest);
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_1"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_2"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_3"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_4"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_5"));
+  EXPECT_FALSE(new_filter_state.hasDataWithName("test_6"));
+}
+
+TEST_F(FilterStateImplTest, HasDataAtOrAboveLifeSpan) {
+  filter_state().setData("test_1", std::make_unique<SimpleType>(1),
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::FilterChain));
+  EXPECT_FALSE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamRequest));
+  EXPECT_FALSE(
+      filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamConnection));
+
+  filter_state().setData("test_2", std::make_unique<SimpleType>(2),
+                         FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::DownstreamRequest);
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::FilterChain));
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamRequest));
+  EXPECT_FALSE(
+      filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamConnection));
+
+  filter_state().setData("test_3", std::make_unique<SimpleType>(3),
+                         FilterState::StateType::ReadOnly,
+                         FilterState::LifeSpan::DownstreamConnection);
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::FilterChain));
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamRequest));
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamConnection));
+}
+
+TEST_F(FilterStateImplTest, SetSameDataWithDifferentLifeSpan) {
+  filter_state().setData("test_1", std::make_unique<SimpleType>(1), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamConnection);
+  // Test reset on smaller LifeSpan
+  EXPECT_THROW_WITH_MESSAGE(
+      filter_state().setData("test_1", std::make_unique<SimpleType>(2),
+                             FilterState::StateType::Mutable, FilterState::LifeSpan::FilterChain),
+      EnvoyException,
+      "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
+  EXPECT_THROW_WITH_MESSAGE(
+      filter_state().setData("test_1", std::make_unique<SimpleType>(2),
+                             FilterState::StateType::Mutable,
+                             FilterState::LifeSpan::DownstreamRequest),
+      EnvoyException,
+      "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
+
+  // Still mutable on the correct LifeSpan.
+  filter_state().setData("test_1", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamConnection);
+  EXPECT_EQ(2, filter_state().getDataMutable<SimpleType>("test_1").access());
+
+  filter_state().setData("test_2", std::make_unique<SimpleType>(1), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamRequest);
+  // Test reset on smaller and greater LifeSpan
+  EXPECT_THROW_WITH_MESSAGE(
+      filter_state().setData("test_2", std::make_unique<SimpleType>(2),
+                             FilterState::StateType::Mutable, FilterState::LifeSpan::FilterChain),
+      EnvoyException,
+      "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
+  EXPECT_THROW_WITH_MESSAGE(
+      filter_state().setData("test_2", std::make_unique<SimpleType>(2),
+                             FilterState::StateType::Mutable,
+                             FilterState::LifeSpan::DownstreamConnection),
+      EnvoyException,
+      "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
+
+  // Still mutable on the correct LifeSpan.
+  filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
+                         FilterState::LifeSpan::DownstreamRequest);
+  EXPECT_EQ(2, filter_state().getDataMutable<SimpleType>("test_2").access());
+}
+
 } // namespace StreamInfo
 } // namespace Envoy
diff --git a/test/common/stream_info/stream_info_impl_test.cc b/test/common/stream_info/stream_info_impl_test.cc
index 0ab04959abf9a..37831316855b8 100644
--- a/test/common/stream_info/stream_info_impl_test.cc
+++ b/test/common/stream_info/stream_info_impl_test.cc
@@ -163,7 +163,8 @@ TEST_F(StreamInfoImplTest, MiscSettersAndGetters) {
     EXPECT_EQ(&route_entry, stream_info.routeEntry());
 
     stream_info.filterState().setData("test", std::make_unique<TestIntAccessor>(1),
-                                      FilterState::StateType::ReadOnly);
+                                      FilterState::StateType::ReadOnly,
+                                      FilterState::LifeSpan::FilterChain);
     EXPECT_EQ(1, stream_info.filterState().getDataReadOnly<TestIntAccessor>("test").access());
 
     EXPECT_EQ("", stream_info.requestedServerName());
diff --git a/test/common/stream_info/test_util.h b/test/common/stream_info/test_util.h
index 1d2354850e6e4..4f153adbdae69 100644
--- a/test/common/stream_info/test_util.h
+++ b/test/common/stream_info/test_util.h
@@ -11,7 +11,7 @@ namespace Envoy {
 
 class TestStreamInfo : public StreamInfo::StreamInfo {
 public:
-  TestStreamInfo() {
+  TestStreamInfo() : filter_state_(Envoy::StreamInfo::FilterState::LifeSpan::FilterChain) {
     tm fake_time;
     memset(&fake_time, 0, sizeof(fake_time));
     fake_time.tm_year = 99; // tm < 1901-12-13 20:45:52 is not valid on macOS
@@ -225,7 +225,7 @@ class TestStreamInfo : public StreamInfo::StreamInfo {
   Ssl::ConnectionInfoConstSharedPtr upstream_connection_info_;
   const Router::RouteEntry* route_entry_{};
   envoy::api::v2::core::Metadata metadata_{};
-  Envoy::StreamInfo::FilterStateImpl filter_state_{};
+  Envoy::StreamInfo::FilterStateImpl filter_state_;
   Envoy::StreamInfo::UpstreamTiming upstream_timing_;
   std::string requested_server_name_;
   std::string upstream_transport_failure_reason_;
diff --git a/test/common/tcp_proxy/tcp_proxy_test.cc b/test/common/tcp_proxy/tcp_proxy_test.cc
index 523724f07961e..81d7096f7e0d3 100644
--- a/test/common/tcp_proxy/tcp_proxy_test.cc
+++ b/test/common/tcp_proxy/tcp_proxy_test.cc
@@ -751,7 +751,8 @@ TEST(ConfigTest, PerConnectionClusterWithTopLevelMetadataMatchConfig) {
   NiceMock<Network::MockConnection> connection;
   connection.stream_info_.filterState().setData(
       "envoy.tcp_proxy.cluster", std::make_unique<PerConnectionCluster>("filter_state_cluster"),
-      StreamInfo::FilterState::StateType::Mutable);
+      StreamInfo::FilterState::StateType::Mutable,
+      StreamInfo::FilterState::LifeSpan::DownstreamConnection);
 
   const auto route = config_obj.getRouteFromEntries(connection);
   EXPECT_NE(nullptr, route);
@@ -1772,7 +1773,8 @@ TEST_F(TcpProxyRoutingTest, DEPRECATED_FEATURE_TEST(UseClusterFromPerConnectionC
   NiceMock<StreamInfo::MockStreamInfo> stream_info;
   stream_info.filterState().setData("envoy.tcp_proxy.cluster",
                                     std::make_unique<PerConnectionCluster>("filter_state_cluster"),
-                                    StreamInfo::FilterState::StateType::Mutable);
+                                    StreamInfo::FilterState::StateType::Mutable,
+                                    StreamInfo::FilterState::LifeSpan::DownstreamConnection);
   ON_CALL(connection_, streamInfo()).WillByDefault(ReturnRef(stream_info));
   EXPECT_CALL(Const(connection_), streamInfo()).WillRepeatedly(ReturnRef(stream_info));
 
@@ -1792,7 +1794,8 @@ TEST_F(TcpProxyRoutingTest, DEPRECATED_FEATURE_TEST(UpstreamServerName)) {
   NiceMock<StreamInfo::MockStreamInfo> stream_info;
   stream_info.filterState().setData("envoy.network.upstream_server_name",
                                     std::make_unique<UpstreamServerName>("www.example.com"),
-                                    StreamInfo::FilterState::StateType::ReadOnly);
+                                    StreamInfo::FilterState::StateType::ReadOnly,
+                                    StreamInfo::FilterState::LifeSpan::DownstreamConnection);
 
   ON_CALL(connection_, streamInfo()).WillByDefault(ReturnRef(stream_info));
   EXPECT_CALL(Const(connection_), streamInfo()).WillRepeatedly(ReturnRef(stream_info));
@@ -1827,7 +1830,8 @@ TEST_F(TcpProxyRoutingTest, DEPRECATED_FEATURE_TEST(ApplicationProtocols)) {
   stream_info.filterState().setData(
       Network::ApplicationProtocols::key(),
       std::make_unique<Network::ApplicationProtocols>(std::vector<std::string>{"foo", "bar"}),
-      StreamInfo::FilterState::StateType::ReadOnly);
+      StreamInfo::FilterState::StateType::ReadOnly,
+      StreamInfo::FilterState::LifeSpan::DownstreamConnection);
 
   ON_CALL(connection_, streamInfo()).WillByDefault(ReturnRef(stream_info));
   EXPECT_CALL(Const(connection_), streamInfo()).WillRepeatedly(ReturnRef(stream_info));
diff --git a/test/extensions/access_loggers/grpc/http_grpc_access_log_impl_test.cc b/test/extensions/access_loggers/grpc/http_grpc_access_log_impl_test.cc
index 11ea5848e7d01..32ccf2641eb55 100644
--- a/test/extensions/access_loggers/grpc/http_grpc_access_log_impl_test.cc
+++ b/test/extensions/access_loggers/grpc/http_grpc_access_log_impl_test.cc
@@ -143,9 +143,11 @@ TEST_F(HttpGrpcAccessLogTest, Marshalling) {
     (*stream_info.metadata_.mutable_filter_metadata())["foo"] = ProtobufWkt::Struct();
     stream_info.filter_state_.setData("string_accessor",
                                       std::make_unique<Router::StringAccessorImpl>("test_value"),
-                                      StreamInfo::FilterState::StateType::ReadOnly);
+                                      StreamInfo::FilterState::StateType::ReadOnly,
+                                      StreamInfo::FilterState::LifeSpan::FilterChain);
     stream_info.filter_state_.setData("serialized", std::make_unique<TestSerializedFilterState>(),
-                                      StreamInfo::FilterState::StateType::ReadOnly);
+                                      StreamInfo::FilterState::StateType::ReadOnly,
+                                      StreamInfo::FilterState::LifeSpan::FilterChain);
     expectLog(R"EOF(
 common_properties:
   downstream_remote_address:
diff --git a/test/extensions/filters/http/jwt_authn/filter_config_test.cc b/test/extensions/filters/http/jwt_authn/filter_config_test.cc
index 1b9583ac3401e..a202d67c147ff 100644
--- a/test/extensions/filters/http/jwt_authn/filter_config_test.cc
+++ b/test/extensions/filters/http/jwt_authn/filter_config_test.cc
@@ -37,7 +37,7 @@ TEST(HttpJwtAuthnFilterConfigTest, FindByMatch) {
   NiceMock<Server::Configuration::MockFactoryContext> context;
   FilterConfig filter_conf(proto_config, "", context);
 
-  StreamInfo::FilterStateImpl filter_state;
+  StreamInfo::FilterStateImpl filter_state(StreamInfo::FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_conf.findVerifier(
                   Http::TestHeaderMapImpl{
                       {":method", "GET"},
@@ -80,26 +80,28 @@ TEST(HttpJwtAuthnFilterConfigTest, FindByFilterState) {
   FilterConfig filter_conf(proto_config, "", context);
 
   // Empty filter_state
-  StreamInfo::FilterStateImpl filter_state1;
+  StreamInfo::FilterStateImpl filter_state1(StreamInfo::FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_conf.findVerifier(Http::TestHeaderMapImpl(), filter_state1) == nullptr);
 
   // Wrong selector
-  StreamInfo::FilterStateImpl filter_state2;
-  filter_state2.setData("jwt_selector",
-                        std::make_unique<Router::StringAccessorImpl>("wrong_selector"),
-                        StreamInfo::FilterState::StateType::ReadOnly);
+  StreamInfo::FilterStateImpl filter_state2(StreamInfo::FilterState::LifeSpan::FilterChain);
+  filter_state2.setData(
+      "jwt_selector", std::make_unique<Router::StringAccessorImpl>("wrong_selector"),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_conf.findVerifier(Http::TestHeaderMapImpl(), filter_state2) == nullptr);
 
   // correct selector
-  StreamInfo::FilterStateImpl filter_state3;
+  StreamInfo::FilterStateImpl filter_state3(StreamInfo::FilterState::LifeSpan::FilterChain);
   filter_state3.setData("jwt_selector", std::make_unique<Router::StringAccessorImpl>("selector1"),
-                        StreamInfo::FilterState::StateType::ReadOnly);
+                        StreamInfo::FilterState::StateType::ReadOnly,
+                        StreamInfo::FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_conf.findVerifier(Http::TestHeaderMapImpl(), filter_state3) != nullptr);
 
   // correct selector
-  StreamInfo::FilterStateImpl filter_state4;
+  StreamInfo::FilterStateImpl filter_state4(StreamInfo::FilterState::LifeSpan::FilterChain);
   filter_state4.setData("jwt_selector", std::make_unique<Router::StringAccessorImpl>("selector2"),
-                        StreamInfo::FilterState::StateType::ReadOnly);
+                        StreamInfo::FilterState::StateType::ReadOnly,
+                        StreamInfo::FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_conf.findVerifier(Http::TestHeaderMapImpl(), filter_state4) != nullptr);
 }
 
diff --git a/test/extensions/filters/http/jwt_authn/filter_integration_test.cc b/test/extensions/filters/http/jwt_authn/filter_integration_test.cc
index 5be4b491b10eb..f5812c904e5fb 100644
--- a/test/extensions/filters/http/jwt_authn/filter_integration_test.cc
+++ b/test/extensions/filters/http/jwt_authn/filter_integration_test.cc
@@ -32,7 +32,8 @@ class HeaderToFilterStateFilter : public Http::PassThroughDecoderFilter {
     if (entry) {
       decoder_callbacks_->streamInfo().filterState().setData(
           state_, std::make_unique<Router::StringAccessorImpl>(entry->value().getStringView()),
-          StreamInfo::FilterState::StateType::ReadOnly);
+          StreamInfo::FilterState::StateType::ReadOnly,
+          StreamInfo::FilterState::LifeSpan::FilterChain);
     }
     return Http::FilterHeadersStatus::Continue;
   }
diff --git a/test/mocks/stream_info/mocks.cc b/test/mocks/stream_info/mocks.cc
index 1f4c4a8f72285..57d2ea436ff9d 100644
--- a/test/mocks/stream_info/mocks.cc
+++ b/test/mocks/stream_info/mocks.cc
@@ -15,7 +15,8 @@ namespace Envoy {
 namespace StreamInfo {
 
 MockStreamInfo::MockStreamInfo()
-    : downstream_local_address_(new Network::Address::Ipv4Instance("127.0.0.2")),
+    : filter_state_(FilterState::LifeSpan::FilterChain),
+      downstream_local_address_(new Network::Address::Ipv4Instance("127.0.0.2")),
       downstream_direct_remote_address_(new Network::Address::Ipv4Instance("127.0.0.1")),
       downstream_remote_address_(new Network::Address::Ipv4Instance("127.0.0.1")) {
   ON_CALL(*this, upstreamHost()).WillByDefault(ReturnPointee(&host_));