diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/HttpMessageParser.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/HttpMessageParser.scala
index 40803632d..0bfd35dfd 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/HttpMessageParser.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/HttpMessageParser.scala
@@ -215,8 +215,11 @@ private[http] trait HttpMessageParser[Output >: MessageOutput <: ParserOutput] {
         case h: Connection => ch match {
             case None =>
               parseHeaderLines(input, lineEnd, headers += h, headerCount + 1, Some(h), clh, cth, isChunked, e100c, hh)
-            case Some(x) => parseHeaderLines(input, lineEnd, headers, headerCount, Some(x.append(h.tokens)), clh, cth,
-                isChunked, e100c, hh)
+            // count each merged Connection header towards the limit: the tokens are accumulated into `x` (an O(n) copy
+            // per header), so without incrementing headerCount the `headerCount < maxHeaderCount` guard never trips and
+            // a flood of Connection headers drives unbounded quadratic work from a single message
+            case Some(x) => parseHeaderLines(input, lineEnd, headers, headerCount + 1, Some(x.append(h.tokens)), clh,
+                cth, isChunked, e100c, hh)
           }
         case h: Host =>
           if (!hh || isResponseParser)
diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/parsing/RequestParserSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/parsing/RequestParserSpec.scala
index a0fb3dc5b..fd9a160e3 100644
--- a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/parsing/RequestParserSpec.scala
+++ b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/parsing/RequestParserSpec.scala
@@ -640,6 +640,28 @@ abstract class RequestParserSpec(mode: String, newLine: String) extends AnyFreeS
           ErrorInfo("HTTP header value exceeds the configured limit of 32 characters"))
       }
 
+      "with more headers than the configured limit" in new Test {
+        override def parserSettings: ParserSettings = super.parserSettings.withMaxHeaderCount(2)
+        """GET / HTTP/1.1
+          |A: 1
+          |B: 2
+          |C: 3""" should parseToError(
+          BadRequest,
+          ErrorInfo("HTTP message contains more than the configured limit of 2 headers"))
+      }
+
+      "with more repeated Connection headers than the configured limit" in new Test {
+        // repeated Connection headers are merged into one, but each still counts towards maxHeaderCount so that a
+        // flood cannot bypass the limit and force unbounded quadratic token accumulation
+        override def parserSettings: ParserSettings = super.parserSettings.withMaxHeaderCount(2)
+        """GET / HTTP/1.1
+          |Connection: a
+          |Connection: b
+          |Connection: c""" should parseToError(
+          BadRequest,
+          ErrorInfo("HTTP message contains more than the configured limit of 2 headers"))
+      }
+
       "with an invalid Content-Length header value" in new Test {
         """GET / HTTP/1.0
           |Content-Length: 1.5