diff --git a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/max-chunk-count.excludes b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/max-chunk-count.excludes
new file mode 100644
index 000000000..9211d2d19
--- /dev/null
+++ b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/max-chunk-count.excludes
@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# new max-chunk-count setting
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.settings.ParserSettings.getMaxChunkCount")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.ParserSettings.maxChunkCount")
diff --git a/http-core/src/main/resources/reference.conf b/http-core/src/main/resources/reference.conf
index b47d6c48e..74bb229ed 100644
--- a/http-core/src/main/resources/reference.conf
+++ b/http-core/src/main/resources/reference.conf
@@ -736,6 +736,7 @@ pekko.http {
     max-header-count           = 64
     max-chunk-ext-length       = 256
     max-chunk-size             = 1m
+    max-chunk-count            = 100000
 
     # HTTP comments (as e.g. prominently used in User-Agent headers) can be nested. To avoid too deep nesting
     # and the associated parsing and storage cost, the depth of nested comments is limited to the given value.
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 2678707b4..110564916 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
@@ -249,7 +249,8 @@ private[http] trait HttpMessageParser[Output >: MessageOutput <: ParserOutput] {
   }
 
   protected final def parseChunk(
-      input: ByteString, offset: Int, isLastMessage: Boolean, totalBytesRead: Long): StateResult = {
+      input: ByteString, offset: Int, isLastMessage: Boolean, totalBytesRead: Long, chunkCount: Int = 0)
+      : StateResult = {
     @tailrec def parseTrailer(extension: String, lineStart: Int, headers: List[HttpHeader] = Nil,
         headerCount: Int = 0): StateResult = {
       var errorInfo: ErrorInfo = null
@@ -277,10 +278,14 @@ private[http] trait HttpMessageParser[Output >: MessageOutput <: ParserOutput] {
 
     def parseChunkBody(chunkSize: Int, extension: String, cursor: Int): StateResult =
       if (chunkSize > 0) {
+        if (chunkCount >= settings.maxChunkCount)
+          failEntityStream(
+            s"HTTP chunk count exceeds the configured limit of ${settings.maxChunkCount} chunks")
         val chunkBodyEnd = cursor + chunkSize
         def result(terminatorLen: Int) = {
           emit(EntityChunk(HttpEntity.Chunk(input.slice(cursor, chunkBodyEnd).compact, extension)))
-          Trampoline(_ => parseChunk(input, chunkBodyEnd + terminatorLen, isLastMessage, totalBytesRead + chunkSize))
+          Trampoline(_ =>
+            parseChunk(input, chunkBodyEnd + terminatorLen, isLastMessage, totalBytesRead + chunkSize, chunkCount + 1))
         }
         byteAt(input, chunkBodyEnd) match {
           case CR_BYTE if byteAt(input, chunkBodyEnd + 1) == LF_BYTE => result(2)
@@ -318,7 +323,8 @@ private[http] trait HttpMessageParser[Output >: MessageOutput <: ParserOutput] {
 
     try parseSize(offset, 0)
     catch {
-      case NotEnoughDataException => continue(input, offset)(parseChunk(_, _, isLastMessage, totalBytesRead))
+      case NotEnoughDataException =>
+        continue(input, offset)(parseChunk(_, _, isLastMessage, totalBytesRead, chunkCount))
     }
   }
 
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ParserSettingsImpl.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ParserSettingsImpl.scala
index aeb3715c9..70885f6f1 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ParserSettingsImpl.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ParserSettingsImpl.scala
@@ -44,6 +44,7 @@ private[pekko] final case class ParserSettingsImpl(
     maxToStrictBytes: Long,
     maxChunkExtLength: Int,
     maxChunkSize: Int,
+    maxChunkCount: Int,
     maxCommentParsingDepth: Int,
     uriParsingMode: Uri.ParsingMode,
     cookieParsingMode: CookieParsingMode,
@@ -71,6 +72,7 @@ private[pekko] final case class ParserSettingsImpl(
   require(maxContentLengthSetting.forall(_ >= 0), "if set max-content-length must be >= 0")
   require(maxChunkExtLength > 0, "max-chunk-ext-length must be > 0")
   require(maxChunkSize > 0, "max-chunk-size must be > 0")
+  require(maxChunkCount > 0, "max-chunk-count must be > 0")
   require(maxCommentParsingDepth > 0, "max-comment-parsing-depth must be > 0")
 
   override val defaultHeaderValueCacheLimit: Int = headerValueCacheLimits("default")
@@ -112,6 +114,7 @@ object ParserSettingsImpl extends SettingsCompanionImpl[ParserSettingsImpl]("pek
       c.getPossiblyInfiniteBytes("max-to-strict-bytes"),
       c.getIntBytes("max-chunk-ext-length"),
       c.getIntBytes("max-chunk-size"),
+      c.getIntBytes("max-chunk-count"),
       c.getInt("max-comment-parsing-depth"),
       Uri.ParsingMode(c.getString("uri-parsing-mode")),
       CookieParsingMode(c.getString("cookie-parsing-mode")),
diff --git a/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ParserSettings.scala b/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ParserSettings.scala
index c726c6597..03df29a10 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ParserSettings.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ParserSettings.scala
@@ -44,6 +44,7 @@ abstract class ParserSettings private[pekko] () extends BodyPartParser.Settings
   def getMaxToStrictBytes: Long
   def getMaxChunkExtLength: Int
   def getMaxChunkSize: Int
+  def getMaxChunkCount: Int
   def getMaxCommentParsingDepth: Int
   def getUriParsingMode: Uri.ParsingMode
   def getCookieParsingMode: ParserSettings.CookieParsingMode
@@ -74,6 +75,7 @@ abstract class ParserSettings private[pekko] () extends BodyPartParser.Settings
   def withMaxToStrictBytes(newValue: Long): ParserSettings = self.copy(maxToStrictBytes = newValue)
   def withMaxChunkExtLength(newValue: Int): ParserSettings = self.copy(maxChunkExtLength = newValue)
   def withMaxChunkSize(newValue: Int): ParserSettings = self.copy(maxChunkSize = newValue)
+  def withMaxChunkCount(newValue: Int): ParserSettings = self.copy(maxChunkCount = newValue)
   def withMaxCommentParsingDepth(newValue: Int): ParserSettings = self.copy(maxCommentParsingDepth = newValue)
   def withUriParsingMode(newValue: Uri.ParsingMode): ParserSettings = self.copy(uriParsingMode = newValue.asScala)
   def withCookieParsingMode(newValue: ParserSettings.CookieParsingMode): ParserSettings =
diff --git a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ParserSettings.scala b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ParserSettings.scala
index 2c0d455e1..23ee3c75c 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ParserSettings.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ParserSettings.scala
@@ -47,6 +47,7 @@ abstract class ParserSettings private[pekko] () extends pekko.http.javadsl.setti
   def maxToStrictBytes: Long
   def maxChunkExtLength: Int
   def maxChunkSize: Int
+  def maxChunkCount: Int
   def maxCommentParsingDepth: Int
   def uriParsingMode: Uri.ParsingMode
   def cookieParsingMode: ParserSettings.CookieParsingMode
@@ -68,6 +69,7 @@ abstract class ParserSettings private[pekko] () extends pekko.http.javadsl.setti
   override def getCookieParsingMode: js.ParserSettings.CookieParsingMode = this.cookieParsingMode
   override def getHeaderValueCacheLimits: util.Map[String, Int] = this.headerValueCacheLimits.asJava
   override def getMaxChunkExtLength = this.maxChunkExtLength
+  override def getMaxChunkCount = this.maxChunkCount
   override def getUriParsingMode: pekko.http.javadsl.model.Uri.ParsingMode = this.uriParsingMode
   override def getMaxHeaderCount = this.maxHeaderCount
   override def getMaxContentLength = this.maxContentLength
@@ -111,6 +113,7 @@ abstract class ParserSettings private[pekko] () extends pekko.http.javadsl.setti
   override def withMaxToStrictBytes(newValue: Long): ParserSettings = self.copy(maxToStrictBytes = newValue)
   override def withMaxChunkExtLength(newValue: Int): ParserSettings = self.copy(maxChunkExtLength = newValue)
   override def withMaxChunkSize(newValue: Int): ParserSettings = self.copy(maxChunkSize = newValue)
+  override def withMaxChunkCount(newValue: Int): ParserSettings = self.copy(maxChunkCount = newValue)
   override def withMaxCommentParsingDepth(newValue: Int): ParserSettings = self.copy(maxCommentParsingDepth = newValue)
   override def withIllegalHeaderWarnings(newValue: Boolean): ParserSettings =
     self.copy(illegalHeaderWarnings = newValue)
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 6782492d3..53ff54e85 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
@@ -556,6 +556,16 @@ abstract class RequestParserSpec(mode: String, newLine: String) extends AnyFreeS
         closeAfterResponseCompletion shouldEqual Seq(false)
       }
 
+      "too many chunks" in new Test {
+        override def parserSettings: ParserSettings =
+          super.parserSettings.withMaxChunkCount(5)
+
+        val chunks = (1 to 10).map(i => s"1\na\n").mkString + "0\n"
+        val result = multiParse(newParser)(Seq(prep(start + chunks)))
+        val errors = result.collect { case Left(EntityStreamError(info)) => info.summary }
+        errors should contain("HTTP chunk count exceeds the configured limit of 5 chunks")
+      }
+
       "an illegal chunk termination" in new Test {
         Seq(
           start,