diff --git a/source/common/http/http1/codec_impl.cc b/source/common/http/http1/codec_impl.cc
index 7312f322cae3c..e5738743fd577 100644
--- a/source/common/http/http1/codec_impl.cc
+++ b/source/common/http/http1/codec_impl.cc
@@ -387,15 +387,23 @@ http_parser_settings ConnectionImpl::settings_{
       return static_cast<ConnectionImpl*>(parser->data)->onHeadersCompleteBase();
     },
     [](http_parser* parser, const char* at, size_t length) -> int {
-      static_cast<ConnectionImpl*>(parser->data)->onBody(at, length);
+      static_cast<ConnectionImpl*>(parser->data)->bufferBody(at, length);
       return 0;
     },
     [](http_parser* parser) -> int {
       static_cast<ConnectionImpl*>(parser->data)->onMessageCompleteBase();
       return 0;
     },
-    nullptr, // on_chunk_header
-    nullptr  // on_chunk_complete
+    [](http_parser* parser) -> int {
+      // A 0-byte chunk header is used to signal the end of the chunked body.
+      // When this function is called, http-parser holds the size of the chunk in
+      // parser->content_length. See
+      // https://github.com/nodejs/http-parser/blob/v2.9.3/http_parser.h#L336
+      const bool is_final_chunk = (parser->content_length == 0);
+      static_cast<ConnectionImpl*>(parser->data)->onChunkHeader(is_final_chunk);
+      return 0;
+    },
+    nullptr // on_chunk_complete
 };
 
 ConnectionImpl::ConnectionImpl(Network::Connection& connection, Stats::Scope& stats,
@@ -451,18 +459,15 @@ bool ConnectionImpl::maybeDirectDispatch(Buffer::Instance& data) {
     return false;
   }
 
-  ssize_t total_parsed = 0;
-  for (const Buffer::RawSlice& slice : data.getRawSlices()) {
-    total_parsed += slice.len_;
-    onBody(static_cast<const char*>(slice.mem_), slice.len_);
-  }
-  ENVOY_CONN_LOG(trace, "direct-dispatched {} bytes", connection_, total_parsed);
-  data.drain(total_parsed);
+  ENVOY_CONN_LOG(trace, "direct-dispatched {} bytes", connection_, data.length());
+  onBody(data);
+  data.drain(data.length());
   return true;
 }
 
 void ConnectionImpl::dispatch(Buffer::Instance& data) {
   ENVOY_CONN_LOG(trace, "parsing {} bytes", connection_, data.length());
+  ASSERT(buffered_body_.length() == 0);
 
   if (maybeDirectDispatch(data)) {
     return;
@@ -475,10 +480,18 @@ void ConnectionImpl::dispatch(Buffer::Instance& data) {
   if (data.length() > 0) {
     for (const Buffer::RawSlice& slice : data.getRawSlices()) {
       total_parsed += dispatchSlice(static_cast<const char*>(slice.mem_), slice.len_);
+      if (HTTP_PARSER_ERRNO(&parser_) != HPE_OK) {
+        // Parse errors trigger an exception in dispatchSlice so we are guaranteed to be paused at
+        // this point.
+        ASSERT(HTTP_PARSER_ERRNO(&parser_) == HPE_PAUSED);
+        break;
+      }
     }
+    dispatchBufferedBody();
   } else {
     dispatchSlice(nullptr, 0);
   }
+  ASSERT(buffered_body_.length() == 0);
 
   ENVOY_CONN_LOG(trace, "parsed {} bytes", connection_, total_parsed);
   data.drain(total_parsed);
@@ -611,9 +624,31 @@ int ConnectionImpl::onHeadersCompleteBase() {
   return handling_upgrade_ ? 2 : rc;
 }
 
+void ConnectionImpl::bufferBody(const char* data, size_t length) {
+  buffered_body_.add(data, length);
+}
+
+void ConnectionImpl::dispatchBufferedBody() {
+  ASSERT(HTTP_PARSER_ERRNO(&parser_) == HPE_OK || HTTP_PARSER_ERRNO(&parser_) == HPE_PAUSED);
+  if (buffered_body_.length() > 0) {
+    onBody(buffered_body_);
+    buffered_body_.drain(buffered_body_.length());
+  }
+}
+
+void ConnectionImpl::onChunkHeader(bool is_final_chunk) {
+  if (is_final_chunk) {
+    // Dispatch body before parsing trailers, so body ends up dispatched even if an error is found
+    // while processing trailers.
+    dispatchBufferedBody();
+  }
+}
+
 void ConnectionImpl::onMessageCompleteBase() {
   ENVOY_CONN_LOG(trace, "message complete", connection_);
 
+  dispatchBufferedBody();
+
   if (handling_upgrade_) {
     // If this is an upgrade request, swallow the onMessageComplete. The
     // upgrade payload will be treated as stream body.
@@ -807,12 +842,11 @@ void ServerConnectionImpl::onUrl(const char* data, size_t length) {
   }
 }
 
-void ServerConnectionImpl::onBody(const char* data, size_t length) {
+void ServerConnectionImpl::onBody(Buffer::Instance& data) {
   ASSERT(!deferred_end_stream_headers_);
   if (active_request_.has_value()) {
-    ENVOY_CONN_LOG(trace, "body size={}", connection_, length);
-    Buffer::OwnedImpl buffer(data, length);
-    active_request_.value().request_decoder_->decodeData(buffer, false);
+    ENVOY_CONN_LOG(trace, "body size={}", connection_, data.length());
+    active_request_.value().request_decoder_->decodeData(data, false);
   }
 }
 
@@ -949,13 +983,11 @@ int ClientConnectionImpl::onHeadersComplete() {
   return cannotHaveBody() ? 1 : 0;
 }
 
-void ClientConnectionImpl::onBody(const char* data, size_t length) {
+void ClientConnectionImpl::onBody(Buffer::Instance& data) {
   ASSERT(!deferred_end_stream_headers_);
   if (pending_response_.has_value()) {
     ASSERT(!pending_response_done_);
-    Buffer::OwnedImpl buffer;
-    buffer.add(data, length);
-    pending_response_.value().decoder_->decodeData(buffer, false);
+    pending_response_.value().decoder_->decodeData(data, false);
   }
 }
 
diff --git a/source/common/http/http1/codec_impl.h b/source/common/http/http1/codec_impl.h
index 60f15e10df10c..c6ebd5e9c9e20 100644
--- a/source/common/http/http1/codec_impl.h
+++ b/source/common/http/http1/codec_impl.h
@@ -246,6 +246,18 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
    */
   size_t dispatchSlice(const char* slice, size_t len);
 
+  /**
+   * Called by the http_parser when body data is received.
+   * @param data supplies the start address.
+   * @param length supplies the length.
+   */
+  void bufferBody(const char* data, size_t length);
+
+  /**
+   * Push the accumulated body through the filter pipeline.
+   */
+  void dispatchBufferedBody();
+
   /**
    * Called when a request/response is beginning. A base routine happens first then a virtual
    * dispatch is invoked.
@@ -284,11 +296,16 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
   virtual int onHeadersComplete() PURE;
 
   /**
-   * Called when body data is received.
-   * @param data supplies the start address.
-   * @param length supplies the length.
+   * Called with body data is available for processing when either:
+   * - There is an accumulated partial body after the parser is done processing bytes read from the
+   * socket
+   * - The parser encounters the last byte of the body
+   * - The codec does a direct dispatch from the read buffer
+   * For performance reasons there is at most one call to onBody per call to HTTP/1
+   * ConnectionImpl::dispatch call.
+   * @param data supplies the body data
    */
-  virtual void onBody(const char* data, size_t length) PURE;
+  virtual void onBody(Buffer::Instance& data) PURE;
 
   /**
    * Called when the request/response is complete.
@@ -296,6 +313,11 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
   void onMessageCompleteBase();
   virtual void onMessageComplete() PURE;
 
+  /**
+   * Called when accepting a chunk header.
+   */
+  void onChunkHeader(bool is_final_chunk);
+
   /**
    * @see onResetStreamBase().
    */
@@ -323,6 +345,10 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
   HeaderParsingState header_parsing_state_{HeaderParsingState::Field};
   HeaderString current_header_field_;
   HeaderString current_header_value_;
+  // Used to accumulate the HTTP message body during the current dispatch call. The accumulated body
+  // is pushed through the filter pipeline either at the end of the current dispatch call, or when
+  // the last byte of the body is processed (whichever happens first).
+  Buffer::OwnedImpl buffered_body_;
   Buffer::WatermarkBuffer output_buffer_;
   Protocol protocol_{Protocol::Http11};
   const uint32_t max_headers_kb_;
@@ -369,7 +395,7 @@ class ServerConnectionImpl : public ServerConnection, public ConnectionImpl {
   void onMessageBegin() override;
   void onUrl(const char* data, size_t length) override;
   int onHeadersComplete() override;
-  void onBody(const char* data, size_t length) override;
+  void onBody(Buffer::Instance& data) override;
   void onMessageComplete() override;
   void onResetStream(StreamResetReason reason) override;
   void sendProtocolError(absl::string_view details) override;
@@ -447,7 +473,7 @@ class ClientConnectionImpl : public ClientConnection, public ConnectionImpl {
   void onMessageBegin() override {}
   void onUrl(const char*, size_t) override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
   int onHeadersComplete() override;
-  void onBody(const char* data, size_t length) override;
+  void onBody(Buffer::Instance& data) override;
   void onMessageComplete() override;
   void onResetStream(StreamResetReason reason) override;
   void sendProtocolError(absl::string_view details) override;
diff --git a/test/common/http/http1/codec_impl_test.cc b/test/common/http/http1/codec_impl_test.cc
index 23aaa9c80f0e2..b483fe563338c 100644
--- a/test/common/http/http1/codec_impl_test.cc
+++ b/test/common/http/http1/codec_impl_test.cc
@@ -18,14 +18,13 @@
 #include "test/test_common/printers.h"
 #include "test/test_common/test_runtime.h"
 
+#include "absl/strings/string_view.h"
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
 using testing::_;
-using testing::AtLeast;
 using testing::InSequence;
 using testing::Invoke;
-using testing::InvokeWithoutArgs;
 using testing::NiceMock;
 using testing::Return;
 using testing::ReturnRef;
@@ -43,6 +42,14 @@ std::string createHeaderFragment(int num_headers) {
   }
   return headers;
 }
+
+Buffer::OwnedImpl createBufferWithOneByteSlices(absl::string_view input) {
+  Buffer::OwnedImpl buffer;
+  for (const char& c : input) {
+    buffer.appendSliceForTest(&c, 1);
+  }
+  return buffer;
+}
 } // namespace
 
 class Http1ServerConnectionImplTest : public testing::Test {
@@ -138,7 +145,7 @@ void Http1ServerConnectionImplTest::expectHeadersTest(Protocol p, bool allow_abs
 
   MockRequestDecoder decoder;
   EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
-  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true));
 
   codec_->dispatch(buffer);
   EXPECT_EQ(0U, buffer.length());
@@ -156,22 +163,26 @@ void Http1ServerConnectionImplTest::expectTrailersTest(bool enable_trailers) {
                                                max_request_headers_kb_, max_request_headers_count_);
   }
 
-  MockRequestDecoder decoder;
+  InSequence sequence;
+  StrictMock<MockRequestDecoder> decoder;
   EXPECT_CALL(callbacks_, newStream(_, _))
       .WillOnce(Invoke([&](ResponseEncoder&, bool) -> RequestDecoder& { return decoder; }));
 
   EXPECT_CALL(decoder, decodeHeaders_(_, false));
 
+  Buffer::OwnedImpl expected_data("Hello World");
   if (enable_trailers) {
-    EXPECT_CALL(decoder, decodeData(_, false)).Times(AtLeast(1));
+    // Verify that body data is delivered before trailers.
+    EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
     EXPECT_CALL(decoder, decodeTrailers_);
   } else {
-    EXPECT_CALL(decoder, decodeData(_, false)).Times(AtLeast(1));
+    EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
     EXPECT_CALL(decoder, decodeData(_, true));
   }
-
-  Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\nb\r\nHello "
-                           "World\r\n0\r\nhello: world\r\nsecond: header\r\n\r\n");
+  Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n"
+                           "6\r\nHello \r\n"
+                           "5\r\nWorld\r\n"
+                           "0\r\nhello: world\r\nsecond: header\r\n\r\n");
   codec_->dispatch(buffer);
   EXPECT_EQ(0U, buffer.length());
 }
@@ -191,9 +202,9 @@ void Http1ServerConnectionImplTest::testTrailersExceedLimit(std::string trailer_
 
   if (enable_trailers) {
     EXPECT_CALL(decoder, decodeHeaders_(_, false));
-    EXPECT_CALL(decoder, decodeData(_, false)).Times(AtLeast(1));
+    EXPECT_CALL(decoder, decodeData(_, false));
   } else {
-    EXPECT_CALL(decoder, decodeData(_, false)).Times(AtLeast(1));
+    EXPECT_CALL(decoder, decodeData(_, false));
     EXPECT_CALL(decoder, decodeData(_, true));
   }
 
@@ -266,7 +277,7 @@ TEST_F(Http1ServerConnectionImplTest, EmptyHeader) {
       {":path", "/"},
       {":method", "GET"},
   };
-  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true));
 
   Buffer::OwnedImpl buffer("GET / HTTP/1.1\r\nTest:\r\nHello: World\r\n\r\n");
   codec_->dispatch(buffer);
@@ -301,6 +312,7 @@ TEST_F(Http1ServerConnectionImplTest, UnsupportedEncoding) {
                             "http/1.1 protocol error: unsupported transfer encoding");
 }
 
+// Verify that data in the two body chunks is merged before the call to decodeData.
 TEST_F(Http1ServerConnectionImplTest, ChunkedBody) {
   initialize();
 
@@ -314,13 +326,82 @@ TEST_F(Http1ServerConnectionImplTest, ChunkedBody) {
       {":method", "POST"},
       {"transfer-encoding", "chunked"},
   };
-  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false));
   Buffer::OwnedImpl expected_data("Hello World");
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false)).Times(1);
-  EXPECT_CALL(decoder, decodeData(_, true)).Times(1);
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
+  // Call to decodeData("", true) happens after.
+  Buffer::OwnedImpl empty("");
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&empty), true));
+
+  Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n"
+                           "6\r\nHello \r\n"
+                           "5\r\nWorld\r\n"
+                           "0\r\n\r\n");
+  codec_->dispatch(buffer);
+  EXPECT_EQ(0U, buffer.length());
+}
 
-  Buffer::OwnedImpl buffer(
-      "POST / HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\nb\r\nHello World\r\n0\r\n\r\n");
+// Verify dispatch behavior when dispatching an incomplete chunk, and resumption of the parse via a
+// second dispatch.
+TEST_F(Http1ServerConnectionImplTest, ChunkedBodySplitOverTwoDispatches) {
+  initialize();
+
+  InSequence sequence;
+
+  MockRequestDecoder decoder;
+  EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
+
+  TestHeaderMapImpl expected_headers{
+      {":path", "/"},
+      {":method", "POST"},
+      {"transfer-encoding", "chunked"},
+  };
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false));
+  Buffer::OwnedImpl expected_data1("Hello Worl");
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data1), false));
+
+  Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n"
+                           "6\r\nHello \r\n"
+                           "5\r\nWorl");
+  codec_->dispatch(buffer);
+  EXPECT_EQ(0U, buffer.length());
+
+  // Process the rest of the body and final chunk.
+  Buffer::OwnedImpl expected_data2("d");
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data2), false));
+  EXPECT_CALL(decoder, decodeData(_, true));
+
+  Buffer::OwnedImpl buffer2("d\r\n"
+                            "0\r\n\r\n");
+  codec_->dispatch(buffer2);
+  EXPECT_EQ(0U, buffer2.length());
+}
+
+// Verify that headers and chunked body are processed correctly and data is merged before the
+// decodeData call even if delivered in a buffer that holds 1 byte per slice.
+TEST_F(Http1ServerConnectionImplTest, ChunkedBodyFragmentedBuffer) {
+  initialize();
+
+  InSequence sequence;
+
+  MockRequestDecoder decoder;
+  EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
+
+  TestHeaderMapImpl expected_headers{
+      {":path", "/"},
+      {":method", "POST"},
+      {"transfer-encoding", "chunked"},
+  };
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false));
+  Buffer::OwnedImpl expected_data("Hello World");
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
+  EXPECT_CALL(decoder, decodeData(_, true));
+
+  Buffer::OwnedImpl buffer =
+      createBufferWithOneByteSlices("POST / HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n"
+                                    "6\r\nHello \r\n"
+                                    "5\r\nWorld\r\n"
+                                    "0\r\n\r\n");
   codec_->dispatch(buffer);
   EXPECT_EQ(0U, buffer.length());
 }
@@ -338,10 +419,10 @@ TEST_F(Http1ServerConnectionImplTest, ChunkedBodyCase) {
       {":method", "POST"},
       {"transfer-encoding", "Chunked"},
   };
-  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false));
   Buffer::OwnedImpl expected_data("Hello World");
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false)).Times(1);
-  EXPECT_CALL(decoder, decodeData(_, true)).Times(1);
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
+  EXPECT_CALL(decoder, decodeData(_, true));
 
   Buffer::OwnedImpl buffer(
       "POST / HTTP/1.1\r\ntransfer-encoding: Chunked\r\n\r\nb\r\nHello World\r\n0\r\n\r\n");
@@ -349,6 +430,32 @@ TEST_F(Http1ServerConnectionImplTest, ChunkedBodyCase) {
   EXPECT_EQ(0U, buffer.length());
 }
 
+// Verify that body dispatch does not happen after detecting a parse error processing a chunk
+// header.
+TEST_F(Http1ServerConnectionImplTest, InvalidChunkHeader) {
+  initialize();
+
+  InSequence sequence;
+
+  MockRequestDecoder decoder;
+  EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
+
+  TestHeaderMapImpl expected_headers{
+      {":path", "/"},
+      {":method", "POST"},
+      {"transfer-encoding", "chunked"},
+  };
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false));
+  EXPECT_CALL(decoder, decodeData(_, _)).Times(0);
+
+  Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n"
+                           "6\r\nHello \r\n"
+                           "invalid\r\nWorl");
+
+  EXPECT_THROW_WITH_MESSAGE(codec_->dispatch(buffer), CodecProtocolException,
+                            "http/1.1 protocol error: HPE_INVALID_CHUNK_SIZE");
+}
+
 TEST_F(Http1ServerConnectionImplTest, IdentityAndChunkedBody) {
   initialize();
 
@@ -389,7 +496,7 @@ TEST_F(Http1ServerConnectionImplTest, Http10) {
   EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
 
   TestHeaderMapImpl expected_headers{{":path", "/"}, {":method", "GET"}};
-  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true));
 
   Buffer::OwnedImpl buffer("GET / HTTP/1.0\r\n\r\n");
   codec_->dispatch(buffer);
@@ -429,7 +536,7 @@ TEST_F(Http1ServerConnectionImplTest, Http10MultipleResponses) {
           return decoder;
         }));
 
-    EXPECT_CALL(decoder, decodeHeaders_(_, true)).Times(1);
+    EXPECT_CALL(decoder, decodeHeaders_(_, true));
     codec_->dispatch(buffer);
 
     std::string output;
@@ -452,7 +559,7 @@ TEST_F(Http1ServerConnectionImplTest, Http10MultipleResponses) {
           response_encoder = &encoder;
           return decoder;
         }));
-    EXPECT_CALL(decoder, decodeHeaders_(_, true)).Times(1);
+    EXPECT_CALL(decoder, decodeHeaders_(_, true));
     codec_->dispatch(buffer);
     EXPECT_EQ(Protocol::Http11, codec_->protocol());
   }
@@ -514,7 +621,10 @@ TEST_F(Http1ServerConnectionImplTest, Http11InvalidTrailerPost) {
       .WillOnce(Invoke([&](ResponseEncoder&, bool) -> RequestDecoder& { return decoder; }));
 
   EXPECT_CALL(decoder, decodeHeaders_(_, false));
-  EXPECT_CALL(decoder, decodeData(_, false)).Times(AtLeast(1));
+  // Verify that body is delivered as soon as the final chunk marker is found, even if an error is
+  // found while processing trailers.
+  Buffer::OwnedImpl expected_data("body");
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
 
   Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\n"
                            "Host: host\r\n"
@@ -584,7 +694,7 @@ TEST_F(Http1ServerConnectionImplTest, SimpleGet) {
   EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
 
   TestHeaderMapImpl expected_headers{{":path", "/"}, {":method", "GET"}};
-  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true));
 
   Buffer::OwnedImpl buffer("GET / HTTP/1.1\r\n\r\n");
   codec_->dispatch(buffer);
@@ -657,7 +767,7 @@ TEST_F(Http1ServerConnectionImplTest, FloodProtection) {
     // In most tests the write output is serialized to a buffer here it is
     // ignored to build up queued "end connection" sentinels.
     EXPECT_CALL(connection_, write(_, _))
-        .Times(1)
+
         .WillOnce(Invoke([&](Buffer::Instance& data, bool) -> void {
           // Move the response out of data while preserving the buffer fragment sentinels.
           local_buffer.move(data);
@@ -708,7 +818,7 @@ TEST_F(Http1ServerConnectionImplTest, FloodProtectionOff) {
     // In most tests the write output is serialized to a buffer here it is
     // ignored to build up queued "end connection" sentinels.
     EXPECT_CALL(connection_, write(_, _))
-        .Times(1)
+
         .WillOnce(Invoke([&](Buffer::Instance& data, bool) -> void {
           // Move the response out of data while preserving the buffer fragment sentinels.
           local_buffer.move(data);
@@ -728,7 +838,7 @@ TEST_F(Http1ServerConnectionImplTest, HostHeaderTranslation) {
   EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
 
   TestHeaderMapImpl expected_headers{{":authority", "hello"}, {":path", "/"}, {":method", "GET"}};
-  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), true));
 
   Buffer::OwnedImpl buffer("GET / HTTP/1.1\r\nHOST: hello\r\n\r\n");
   codec_->dispatch(buffer);
@@ -892,19 +1002,44 @@ TEST_F(Http1ServerConnectionImplTest, PostWithContentLength) {
   EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
 
   TestHeaderMapImpl expected_headers{{"content-length", "5"}, {":path", "/"}, {":method", "POST"}};
-  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false));
 
   Buffer::OwnedImpl expected_data1("12345");
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data1), false)).Times(1);
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data1), false));
 
   Buffer::OwnedImpl expected_data2;
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data2), true)).Times(1);
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data2), true));
 
   Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\ncontent-length: 5\r\n\r\n12345");
   codec_->dispatch(buffer);
   EXPECT_EQ(0U, buffer.length());
 }
 
+// Verify that headers and body with content length are processed correctly and data is merged
+// before the decodeData call even if delivered in a buffer that holds 1 byte per slice.
+TEST_F(Http1ServerConnectionImplTest, PostWithContentLengthFragmentedBuffer) {
+  initialize();
+
+  InSequence sequence;
+
+  MockRequestDecoder decoder;
+  EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
+
+  TestHeaderMapImpl expected_headers{{"content-length", "5"}, {":path", "/"}, {":method", "POST"}};
+  EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_headers), false));
+
+  Buffer::OwnedImpl expected_data1("12345");
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data1), false));
+
+  Buffer::OwnedImpl expected_data2;
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data2), true));
+
+  Buffer::OwnedImpl buffer =
+      createBufferWithOneByteSlices("POST / HTTP/1.1\r\ncontent-length: 5\r\n\r\n12345");
+  codec_->dispatch(buffer);
+  EXPECT_EQ(0U, buffer.length());
+}
+
 TEST_F(Http1ServerConnectionImplTest, HeaderOnlyResponse) {
   initialize();
 
@@ -1268,19 +1403,19 @@ TEST_F(Http1ServerConnectionImplTest, UpgradeRequest) {
   NiceMock<MockRequestDecoder> decoder;
   EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
 
-  EXPECT_CALL(decoder, decodeHeaders_(_, false)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(_, false));
   Buffer::OwnedImpl buffer(
       "POST / HTTP/1.1\r\nConnection: upgrade\r\nUpgrade: foo\r\ncontent-length:5\r\n\r\n");
   codec_->dispatch(buffer);
 
   Buffer::OwnedImpl expected_data1("12345");
   Buffer::OwnedImpl body("12345");
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data1), false)).Times(1);
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data1), false));
   codec_->dispatch(body);
 
   Buffer::OwnedImpl expected_data2("abcd");
   Buffer::OwnedImpl websocket_payload("abcd");
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data2), false)).Times(1);
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data2), false));
   codec_->dispatch(websocket_payload);
 }
 
@@ -1292,8 +1427,8 @@ TEST_F(Http1ServerConnectionImplTest, UpgradeRequestWithEarlyData) {
   EXPECT_CALL(callbacks_, newStream(_, _)).WillOnce(ReturnRef(decoder));
 
   Buffer::OwnedImpl expected_data("12345abcd");
-  EXPECT_CALL(decoder, decodeHeaders_(_, false)).Times(1);
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(_, false));
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
   Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\nConnection: upgrade\r\nUpgrade: "
                            "foo\r\ncontent-length:5\r\n\r\n12345abcd");
   codec_->dispatch(buffer);
@@ -1309,8 +1444,8 @@ TEST_F(Http1ServerConnectionImplTest, UpgradeRequestWithTEChunked) {
   // Even with T-E chunked, the data should neither be inspected for (the not
   // present in this unit test) chunks, but simply passed through.
   Buffer::OwnedImpl expected_data("12345abcd");
-  EXPECT_CALL(decoder, decodeHeaders_(_, false)).Times(1);
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(_, false));
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
   Buffer::OwnedImpl buffer("POST / HTTP/1.1\r\nConnection: upgrade\r\nUpgrade: "
                            "foo\r\ntransfer-encoding: chunked\r\n\r\n12345abcd");
   codec_->dispatch(buffer);
@@ -1326,15 +1461,15 @@ TEST_F(Http1ServerConnectionImplTest, UpgradeRequestWithNoBody) {
   // Make sure we avoid the deferred_end_stream_headers_ optimization for
   // requests-with-no-body.
   Buffer::OwnedImpl expected_data("abcd");
-  EXPECT_CALL(decoder, decodeHeaders_(_, false)).Times(1);
-  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false)).Times(1);
+  EXPECT_CALL(decoder, decodeHeaders_(_, false));
+  EXPECT_CALL(decoder, decodeData(BufferEqual(&expected_data), false));
   Buffer::OwnedImpl buffer(
       "GET / HTTP/1.1\r\nConnection: upgrade\r\nUpgrade: foo\r\ncontent-length: 0\r\n\r\nabcd");
   codec_->dispatch(buffer);
 }
 
 TEST_F(Http1ServerConnectionImplTest, WatermarkTest) {
-  EXPECT_CALL(connection_, bufferLimit()).Times(1).WillOnce(Return(10));
+  EXPECT_CALL(connection_, bufferLimit()).WillOnce(Return(10));
   initialize();
 
   NiceMock<MockRequestDecoder> decoder;
@@ -1630,13 +1765,13 @@ TEST_F(Http1ClientConnectionImplTest, UpgradeResponse) {
   // Send body payload
   Buffer::OwnedImpl expected_data1("12345");
   Buffer::OwnedImpl body("12345");
-  EXPECT_CALL(response_decoder, decodeData(BufferEqual(&expected_data1), false)).Times(1);
+  EXPECT_CALL(response_decoder, decodeData(BufferEqual(&expected_data1), false));
   codec_->dispatch(body);
 
   // Send websocket payload
   Buffer::OwnedImpl expected_data2("abcd");
   Buffer::OwnedImpl websocket_payload("abcd");
-  EXPECT_CALL(response_decoder, decodeData(BufferEqual(&expected_data2), false)).Times(1);
+  EXPECT_CALL(response_decoder, decodeData(BufferEqual(&expected_data2), false));
   codec_->dispatch(websocket_payload);
 }
 
@@ -1655,14 +1790,14 @@ TEST_F(Http1ClientConnectionImplTest, UpgradeResponseWithEarlyData) {
   // Send upgrade headers
   EXPECT_CALL(response_decoder, decodeHeaders_(_, false));
   Buffer::OwnedImpl expected_data("12345abcd");
-  EXPECT_CALL(response_decoder, decodeData(BufferEqual(&expected_data), false)).Times(1);
+  EXPECT_CALL(response_decoder, decodeData(BufferEqual(&expected_data), false));
   Buffer::OwnedImpl response("HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: "
                              "upgrade\r\nUpgrade: websocket\r\n\r\n12345abcd");
   codec_->dispatch(response);
 }
 
 TEST_F(Http1ClientConnectionImplTest, WatermarkTest) {
-  EXPECT_CALL(connection_, bufferLimit()).Times(1).WillOnce(Return(10));
+  EXPECT_CALL(connection_, bufferLimit()).WillOnce(Return(10));
   initialize();
 
   InSequence s;
diff --git a/test/extensions/transport_sockets/alts/alts_integration_test.cc b/test/extensions/transport_sockets/alts/alts_integration_test.cc
index 39189c8170168..587fd3b8e490b 100644
--- a/test/extensions/transport_sockets/alts/alts_integration_test.cc
+++ b/test/extensions/transport_sockets/alts/alts_integration_test.cc
@@ -165,7 +165,7 @@ TEST_P(AltsIntegrationTestValidPeer, RouterRequestAndResponseWithBodyNoBuffer) {
   ConnectionCreationFunction creator = [this]() -> Network::ClientConnectionPtr {
     return makeAltsConnection();
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 class AltsIntegrationTestEmptyPeer : public AltsIntegrationTestBase {
@@ -186,7 +186,7 @@ TEST_P(AltsIntegrationTestEmptyPeer, RouterRequestAndResponseWithBodyNoBuffer) {
   ConnectionCreationFunction creator = [this]() -> Network::ClientConnectionPtr {
     return makeAltsConnection();
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 class AltsIntegrationTestClientInvalidPeer : public AltsIntegrationTestBase {
diff --git a/test/extensions/transport_sockets/tls/integration/ssl_integration_test.cc b/test/extensions/transport_sockets/tls/integration/ssl_integration_test.cc
index bf55f47d034e8..bd755736ef457 100644
--- a/test/extensions/transport_sockets/tls/integration/ssl_integration_test.cc
+++ b/test/extensions/transport_sockets/tls/integration/ssl_integration_test.cc
@@ -89,7 +89,7 @@ TEST_P(SslIntegrationTest, RouterRequestAndResponseWithGiantBodyBuffer) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection({});
   };
-  testRouterRequestAndResponseWithBody(16 * 1024 * 1024, 16 * 1024 * 1024, false, &creator);
+  testRouterRequestAndResponseWithBody(16 * 1024 * 1024, 16 * 1024 * 1024, false, false, &creator);
   checkStats();
 }
 
@@ -97,7 +97,7 @@ TEST_P(SslIntegrationTest, RouterRequestAndResponseWithBodyNoBuffer) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection({});
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -108,7 +108,7 @@ TEST_P(SslIntegrationTest, RouterRequestAndResponseWithBodyNoBufferHttp2) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection(ClientSslTransportOptions().setAlpn(true));
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -116,7 +116,7 @@ TEST_P(SslIntegrationTest, RouterRequestAndResponseWithBodyNoBufferVerifySAN) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection(ClientSslTransportOptions().setSan(true));
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -125,7 +125,7 @@ TEST_P(SslIntegrationTest, RouterRequestAndResponseWithBodyNoBufferHttp2VerifySA
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection(ClientSslTransportOptions().setAlpn(true).setSan(true));
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -244,7 +244,7 @@ TEST_P(SslCertficateIntegrationTest, ServerRsa) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection({});
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -255,7 +255,7 @@ TEST_P(SslCertficateIntegrationTest, ServerEcdsa) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection({});
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -266,7 +266,7 @@ TEST_P(SslCertficateIntegrationTest, ServerRsaEcdsa) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection({});
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -277,7 +277,7 @@ TEST_P(SslCertficateIntegrationTest, ClientRsaOnly) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection(rsaOnlyClientOptions());
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -302,7 +302,7 @@ TEST_P(SslCertficateIntegrationTest, ServerRsaEcdsaClientRsaOnly) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection(rsaOnlyClientOptions());
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -329,7 +329,7 @@ TEST_P(SslCertficateIntegrationTest, ServerEcdsaClientEcdsaOnly) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection(ecdsaOnlyClientOptions());
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -341,7 +341,7 @@ TEST_P(SslCertficateIntegrationTest, ServerRsaEcdsaClientEcdsaOnly) {
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection(ecdsaOnlyClientOptions());
   };
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
 }
 
@@ -550,7 +550,7 @@ TEST_P(SslTapIntegrationTest, RequestWithTextProto) {
     return makeSslClientConnection({});
   };
   const uint64_t id = Network::ConnectionImpl::nextGlobalIdForTest() + 1;
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   checkStats();
   codec_client_->close();
   test_server_->waitForCounterGe("http.config_test.downstream_cx_destroy", 1);
@@ -576,7 +576,7 @@ TEST_P(SslTapIntegrationTest, RequestWithJsonBodyAsStringUpstreamTap) {
     return makeSslClientConnection({});
   };
   const uint64_t id = Network::ConnectionImpl::nextGlobalIdForTest() + 2;
-  testRouterRequestAndResponseWithBody(512, 1024, false, &creator);
+  testRouterRequestAndResponseWithBody(512, 1024, false, false, &creator);
   checkStats();
   codec_client_->close();
   test_server_->waitForCounterGe("http.config_test.downstream_cx_destroy", 1);
diff --git a/test/integration/http2_integration_test.cc b/test/integration/http2_integration_test.cc
index 27d58e6acc84a..5064e051053b4 100644
--- a/test/integration/http2_integration_test.cc
+++ b/test/integration/http2_integration_test.cc
@@ -28,12 +28,41 @@ INSTANTIATE_TEST_SUITE_P(IpVersions, Http2IntegrationTest,
                          TestUtility::ipTestParamsToString);
 
 TEST_P(Http2IntegrationTest, RouterRequestAndResponseWithBodyNoBuffer) {
-  testRouterRequestAndResponseWithBody(1024, 512, false);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false);
+}
+
+TEST_P(Http2IntegrationTest, RouterRequestAndResponseWithGiantBodyNoBuffer) {
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, false);
 }
 
 TEST_P(Http2IntegrationTest, FlowControlOnAndGiantBody) {
   config_helper_.setBufferLimits(1024, 1024); // Set buffer limits upstream and downstream.
-  testRouterRequestAndResponseWithBody(1024 * 1024, 1024 * 1024, false);
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, false);
+}
+
+TEST_P(Http2IntegrationTest, LargeFlowControlOnAndGiantBody) {
+  config_helper_.setBufferLimits(128 * 1024,
+                                 128 * 1024); // Set buffer limits upstream and downstream.
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, false);
+}
+
+TEST_P(Http2IntegrationTest, RouterRequestAndResponseWithBodyAndContentLengthNoBuffer) {
+  testRouterRequestAndResponseWithBody(1024, 512, false, true);
+}
+
+TEST_P(Http2IntegrationTest, RouterRequestAndResponseWithGiantBodyAndContentLengthNoBuffer) {
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, true);
+}
+
+TEST_P(Http2IntegrationTest, FlowControlOnAndGiantBodyWithContentLength) {
+  config_helper_.setBufferLimits(1024, 1024); // Set buffer limits upstream and downstream.
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, true);
+}
+
+TEST_P(Http2IntegrationTest, LargeFlowControlOnAndGiantBodyWithContentLength) {
+  config_helper_.setBufferLimits(128 * 1024,
+                                 128 * 1024); // Set buffer limits upstream and downstream.
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, true);
 }
 
 TEST_P(Http2IntegrationTest, RouterHeaderOnlyRequestAndResponseNoBuffer) {
diff --git a/test/integration/http_integration.cc b/test/integration/http_integration.cc
index 4fe0619165368..f485f3083c640 100644
--- a/test/integration/http_integration.cc
+++ b/test/integration/http_integration.cc
@@ -398,7 +398,7 @@ void HttpIntegrationTest::checkSimpleRequestSuccess(uint64_t expected_request_si
 }
 
 void HttpIntegrationTest::testRouterRequestAndResponseWithBody(
-    uint64_t request_size, uint64_t response_size, bool big_header,
+    uint64_t request_size, uint64_t response_size, bool big_header, bool set_content_length_header,
     ConnectionCreationFunction* create_connection) {
   initialize();
   codec_client_ = makeHttpConnection(
@@ -406,11 +406,16 @@ void HttpIntegrationTest::testRouterRequestAndResponseWithBody(
   Http::TestRequestHeaderMapImpl request_headers{
       {":method", "POST"},    {":path", "/test/long/url"}, {":scheme", "http"},
       {":authority", "host"}, {"x-lyft-user-id", "123"},   {"x-forwarded-for", "10.0.0.1"}};
+  Http::TestResponseHeaderMapImpl response_headers{{":status", "200"}};
+  if (set_content_length_header) {
+    request_headers.setContentLength(request_size);
+    response_headers.setContentLength(response_size);
+  }
   if (big_header) {
     request_headers.addCopy("big", std::string(4096, 'a'));
   }
-  auto response = sendRequestAndWaitForResponse(request_headers, request_size,
-                                                default_response_headers_, response_size);
+  auto response =
+      sendRequestAndWaitForResponse(request_headers, request_size, response_headers, response_size);
   checkSimpleRequestSuccess(request_size, response_size, response.get());
 }
 
diff --git a/test/integration/http_integration.h b/test/integration/http_integration.h
index ec9edb1e73102..9170f391736a2 100644
--- a/test/integration/http_integration.h
+++ b/test/integration/http_integration.h
@@ -173,7 +173,7 @@ class HttpIntegrationTest : public BaseIntegrationTest {
   void testRouterVirtualClusters();
 
   void testRouterRequestAndResponseWithBody(uint64_t request_size, uint64_t response_size,
-                                            bool big_header,
+                                            bool big_header, bool set_content_length_header = false,
                                             ConnectionCreationFunction* creator = nullptr);
   void testRouterHeaderOnlyRequestAndResponse(ConnectionCreationFunction* creator = nullptr,
                                               int upstream_index = 0,
diff --git a/test/integration/idle_timeout_integration_test.cc b/test/integration/idle_timeout_integration_test.cc
index d4827b0d7148f..299d875247bee 100644
--- a/test/integration/idle_timeout_integration_test.cc
+++ b/test/integration/idle_timeout_integration_test.cc
@@ -282,7 +282,7 @@ TEST_P(IdleTimeoutIntegrationTest, PerStreamIdleTimeoutAfterBidiData) {
 // Successful request/response when per-stream idle timeout is configured.
 TEST_P(IdleTimeoutIntegrationTest, PerStreamIdleTimeoutRequestAndResponse) {
   enable_per_stream_idle_timeout_ = true;
-  testRouterRequestAndResponseWithBody(1024, 1024, false, nullptr);
+  testRouterRequestAndResponseWithBody(1024, 1024, false);
 }
 
 TEST_P(IdleTimeoutIntegrationTest, RequestTimeoutConfiguredRequestResponse) {
@@ -292,7 +292,7 @@ TEST_P(IdleTimeoutIntegrationTest, RequestTimeoutConfiguredRequestResponse) {
 
 TEST_P(IdleTimeoutIntegrationTest, RequestTimeoutConfiguredRequestResponseWithBody) {
   enable_request_timeout_ = true;
-  testRouterRequestAndResponseWithBody(1024, 1024, false, nullptr);
+  testRouterRequestAndResponseWithBody(1024, 1024, false);
 }
 
 TEST_P(IdleTimeoutIntegrationTest, RequestTimeoutTriggersOnBodilessPost) {
diff --git a/test/integration/integration_test.cc b/test/integration/integration_test.cc
index 92908225fec6b..424149135ab68 100644
--- a/test/integration/integration_test.cc
+++ b/test/integration/integration_test.cc
@@ -158,12 +158,39 @@ TEST_P(IntegrationTest, ConnectionClose) {
 }
 
 TEST_P(IntegrationTest, RouterRequestAndResponseWithBodyNoBuffer) {
-  testRouterRequestAndResponseWithBody(1024, 512, false);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false);
+}
+
+TEST_P(IntegrationTest, RouterRequestAndResponseWithGiantBodyNoBuffer) {
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, false);
 }
 
 TEST_P(IntegrationTest, FlowControlOnAndGiantBody) {
   config_helper_.setBufferLimits(1024, 1024);
-  testRouterRequestAndResponseWithBody(1024 * 1024, 1024 * 1024, false);
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, false);
+}
+
+TEST_P(IntegrationTest, LargeFlowControlOnAndGiantBody) {
+  config_helper_.setBufferLimits(128 * 1024, 128 * 1024);
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, false);
+}
+
+TEST_P(IntegrationTest, RouterRequestAndResponseWithBodyAndContentLengthNoBuffer) {
+  testRouterRequestAndResponseWithBody(1024, 512, false, true);
+}
+
+TEST_P(IntegrationTest, RouterRequestAndResponseWithGiantBodyAndContentLengthNoBuffer) {
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, true);
+}
+
+TEST_P(IntegrationTest, FlowControlOnAndGiantBodyWithContentLength) {
+  config_helper_.setBufferLimits(1024, 1024);
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, true);
+}
+
+TEST_P(IntegrationTest, LargeFlowControlOnAndGiantBodyWithContentLength) {
+  config_helper_.setBufferLimits(128 * 1024, 128 * 1024);
+  testRouterRequestAndResponseWithBody(10 * 1024 * 1024, 10 * 1024 * 1024, false, true);
 }
 
 TEST_P(IntegrationTest, RouterRequestAndResponseLargeHeaderNoBuffer) {
diff --git a/test/integration/proxy_proto_integration_test.cc b/test/integration/proxy_proto_integration_test.cc
index c10323e75737c..ec62a8991a6ad 100644
--- a/test/integration/proxy_proto_integration_test.cc
+++ b/test/integration/proxy_proto_integration_test.cc
@@ -26,7 +26,7 @@ TEST_P(ProxyProtoIntegrationTest, V1RouterRequestAndResponseWithBodyNoBuffer) {
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 TEST_P(ProxyProtoIntegrationTest, V2RouterRequestAndResponseWithBodyNoBuffer) {
@@ -40,7 +40,7 @@ TEST_P(ProxyProtoIntegrationTest, V2RouterRequestAndResponseWithBodyNoBuffer) {
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 TEST_P(ProxyProtoIntegrationTest, V1RouterRequestAndResponseWithBodyNoBufferV6) {
@@ -51,7 +51,7 @@ TEST_P(ProxyProtoIntegrationTest, V1RouterRequestAndResponseWithBodyNoBufferV6)
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 TEST_P(ProxyProtoIntegrationTest, V2RouterRequestAndResponseWithBodyNoBufferV6) {
@@ -67,7 +67,7 @@ TEST_P(ProxyProtoIntegrationTest, V2RouterRequestAndResponseWithBodyNoBufferV6)
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 TEST_P(ProxyProtoIntegrationTest, RouterProxyUnknownRequestAndResponseWithBodyNoBuffer) {
@@ -78,7 +78,7 @@ TEST_P(ProxyProtoIntegrationTest, RouterProxyUnknownRequestAndResponseWithBodyNo
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 TEST_P(ProxyProtoIntegrationTest, RouterProxyUnknownLongRequestAndResponseWithBodyNoBuffer) {
@@ -89,7 +89,7 @@ TEST_P(ProxyProtoIntegrationTest, RouterProxyUnknownLongRequestAndResponseWithBo
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 // Test that %DOWNSTREAM_DIRECT_REMOTE_ADDRESS%/%DOWNSTREAM_DIRECT_REMOTE_ADDRESS_WITHOUT_PORT%
@@ -110,7 +110,7 @@ TEST_P(ProxyProtoIntegrationTest, AccessLog) {
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
   const std::string log_line = waitForAccessLog(access_log_name_);
   const std::vector<absl::string_view> tokens = StringUtil::splitToken(log_line, " ");
 
@@ -147,7 +147,7 @@ TEST_P(ProxyProtoIntegrationTest, DEPRECATED_FEATURE_TEST(OriginalDst)) {
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 TEST_P(ProxyProtoIntegrationTest, ClusterProvided) {
@@ -177,7 +177,7 @@ TEST_P(ProxyProtoIntegrationTest, ClusterProvided) {
     return conn;
   };
 
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 } // namespace Envoy
diff --git a/test/integration/sds_static_integration_test.cc b/test/integration/sds_static_integration_test.cc
index 8b488b0f58e19..f0a92b8e19c74 100644
--- a/test/integration/sds_static_integration_test.cc
+++ b/test/integration/sds_static_integration_test.cc
@@ -104,7 +104,7 @@ TEST_P(SdsStaticDownstreamIntegrationTest, RouterRequestAndResponseWithGiantBody
   ConnectionCreationFunction creator = [&]() -> Network::ClientConnectionPtr {
     return makeSslClientConnection();
   };
-  testRouterRequestAndResponseWithBody(16 * 1024 * 1024, 16 * 1024 * 1024, false, &creator);
+  testRouterRequestAndResponseWithBody(16 * 1024 * 1024, 16 * 1024 * 1024, false, false, &creator);
 }
 
 class SdsStaticUpstreamIntegrationTest : public testing::TestWithParam<Network::Address::IpVersion>,
@@ -160,7 +160,7 @@ INSTANTIATE_TEST_SUITE_P(IpVersions, SdsStaticUpstreamIntegrationTest,
                          TestUtility::ipTestParamsToString);
 
 TEST_P(SdsStaticUpstreamIntegrationTest, RouterRequestAndResponseWithGiantBodyBuffer) {
-  testRouterRequestAndResponseWithBody(16 * 1024 * 1024, 16 * 1024 * 1024, false, nullptr);
+  testRouterRequestAndResponseWithBody(16 * 1024 * 1024, 16 * 1024 * 1024, false, false, nullptr);
 }
 
 } // namespace Ssl
diff --git a/test/integration/uds_integration_test.cc b/test/integration/uds_integration_test.cc
index ddb9abc68ad6d..e1e63b8fbfb55 100644
--- a/test/integration/uds_integration_test.cc
+++ b/test/integration/uds_integration_test.cc
@@ -111,7 +111,7 @@ TEST_P(UdsListenerIntegrationTest, TestPeerCredentials) {
 
 TEST_P(UdsListenerIntegrationTest, RouterRequestAndResponseWithBodyNoBuffer) {
   ConnectionCreationFunction creator = createConnectionFn();
-  testRouterRequestAndResponseWithBody(1024, 512, false, &creator);
+  testRouterRequestAndResponseWithBody(1024, 512, false, false, &creator);
 }
 
 TEST_P(UdsListenerIntegrationTest, RouterHeaderOnlyRequestAndResponse) {