diff --git a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/max-part-count.excludes b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/max-part-count.excludes
new file mode 100644
index 000000000..4ab0e4924
--- /dev/null
+++ b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/max-part-count.excludes
@@ -0,0 +1,21 @@
+# 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-part-count setting
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.impl.engine.parsing.BodyPartParser#Settings.maxPartCount")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.settings.ParserSettings.getMaxPartCount")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.ParserSettings.maxPartCount")
diff --git a/http-core/src/main/resources/reference.conf b/http-core/src/main/resources/reference.conf
index 0767de86b..19df08f68 100644
--- a/http-core/src/main/resources/reference.conf
+++ b/http-core/src/main/resources/reference.conf
@@ -769,6 +769,15 @@ pekko.http {
     max-chunk-size             = 1m
     max-chunk-count            = 100000
 
+    # The maximum number of body parts a multipart entity may consist of. Each part costs a set of parsed headers and
+    # an entity of its own, so a body packed with minimal parts amplifies the work and allocation a request of a given
+    # size causes, well beyond what max-content-length alone bounds: a body of max-content-length bytes made of
+    # minimal parts runs to well over a hundred thousand of them.
+    #
+    # The default is deliberately generous, chosen to stay above the largest part count the test suite exercises
+    # rather than to be the tightest useful bound. Applications that know their forms are small can set it far lower.
+    max-part-count             = 10000
+
     # 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.
     max-comment-parsing-depth  = 5
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/BodyPartParser.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/BodyPartParser.scala
index 3ee26ed8e..fefb0fa5f 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/BodyPartParser.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/BodyPartParser.scala
@@ -67,6 +67,7 @@ private[http] final class BodyPartParser(
       private var output = collection.immutable.Queue.empty[Output] // FIXME this probably is too wasteful
       private var state: ByteString => StateResult = tryParseInitialBoundary
       private var shouldTerminate = false
+      private var partCount = 0
       // Will be override at the beginning of the parsing (tryParseInitialBoundary and parsePreamble)
       // But initially defined here as norm version to avoid NPE
       private var eolConfiguration: EndOfLineConfiguration = UndefinedEndOfLineConfiguration(boundary)
@@ -129,7 +130,8 @@ private[http] final class BodyPartParser(
           eolConfiguration = eolConfiguration.defineOnce(input)
           if (eolConfiguration.isBoundary(input, 0)) {
             val ix = eolConfiguration.boundaryLength
-            if (eolConfiguration.isEndOfLine(input, ix)) parseHeaderLines(input, ix + eolConfiguration.eolLength)
+            if (eolConfiguration.isEndOfLine(input, ix))
+              parsePartHeaderLines(input, ix + eolConfiguration.eolLength)
             else if (doubleDash(input, ix)) setShouldTerminate()
             else parsePreamble(input)
           } else parsePreamble(input)
@@ -142,7 +144,7 @@ private[http] final class BodyPartParser(
           @tailrec def rec(index: Int): StateResult = {
             val needleEnd = eolConfiguration.boyerMoore.nextIndex(input, index) + eolConfiguration.needle.length
             if (eolConfiguration.isEndOfLine(input, needleEnd))
-              parseHeaderLines(input, needleEnd + eolConfiguration.eolLength)
+              parsePartHeaderLines(input, needleEnd + eolConfiguration.eolLength)
             else if (doubleDash(input, needleEnd)) setShouldTerminate()
             else rec(needleEnd)
           }
@@ -152,6 +154,24 @@ private[http] final class BodyPartParser(
           case NotEnoughDataException => continue(input, 0)((newInput, _) => parsePreamble(newInput))
         }
 
+      /**
+       * Registers the start of a new body part, bounding how many of them one entity may contain. Each part costs a
+       * set of parsed headers and an entity of its own, so a body packed with minimal parts amplifies the work a
+       * request of a given size causes beyond what `max-content-length` bounds. Returns false once the limit is
+       * exhausted, in which case the caller must fail the entity via `failMaxPartCount`.
+       */
+      def startPart(): Boolean = {
+        val withinLimit = partCount < maxPartCount
+        if (withinLimit) partCount += 1
+        withinLimit
+      }
+
+      def failMaxPartCount(): StateResult =
+        fail(s"multipart entity contains more than the configured limit of $maxPartCount parts")
+
+      def parsePartHeaderLines(input: ByteString, lineStart: Int): StateResult =
+        if (startPart()) parseHeaderLines(input, lineStart) else failMaxPartCount()
+
       @tailrec def parseHeaderLines(input: ByteString, lineStart: Int,
           headers: ListBuffer[HttpHeader] = ListBuffer[HttpHeader](),
           headerCount: Int = 0, cth: Option[`Content-Type`] = None): StateResult = {
@@ -177,9 +197,14 @@ private[http] final class BodyPartParser(
           case BoundaryHeader =>
             emit(BodyPartStart(headers.toList, _ => HttpEntity.empty(contentType)))
             val ix = lineStart + eolConfiguration.boundaryLength
-            if (eolConfiguration.isEndOfLine(input, ix))
-              parseHeaderLines(input, ix + eolConfiguration.eolLength, headers, headerCount, None)
-            else if (doubleDash(input, ix)) setShouldTerminate()
+            if (eolConfiguration.isEndOfLine(input, ix)) {
+              // an empty part; the boundary starts another one, so it counts towards the limit as well. We must not
+              // route this through `parsePartHeaderLines`: the self-recursive call below is what keeps this method
+              // tail-recursive, and a mutual recursion here would risk the stack overflow the trampoline in
+              // `parseEntity` guards against.
+              if (startPart()) parseHeaderLines(input, ix + eolConfiguration.eolLength, headers, headerCount, None)
+              else failMaxPartCount()
+            } else if (doubleDash(input, ix)) setShouldTerminate()
             else fail("Illegal multipart boundary in message content")
 
           case EmptyHeader => parseEntity(headers.toList, contentType)(input, lineEnd)
@@ -229,7 +254,7 @@ private[http] final class BodyPartParser(
               // Need to trampoline here, otherwise we have a mutual tail recursion between parseHeaderLines and
               // parseEntity that is not tail-call optimized away and may lead to stack overflows on big chunks of data
               // containing many parts.
-              trampoline(parseHeaderLines(input, needleEnd + eolConfiguration.eolLength))
+              trampoline(parsePartHeaderLines(input, needleEnd + eolConfiguration.eolLength))
             } else if (doubleDash(input, needleEnd)) {
               emitFinalChunk()
               setShouldTerminate()
@@ -309,6 +334,7 @@ private[http] object BodyPartParser {
 
   abstract class Settings extends HttpHeaderParser.Settings {
     def maxHeaderCount: Int
+    def maxPartCount: Int
     def illegalHeaderWarnings: Boolean
     def defaultHeaderValueCacheLimit: Int
   }
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 70885f6f1..f0e7ae5ef 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
@@ -45,6 +45,7 @@ private[pekko] final case class ParserSettingsImpl(
     maxChunkExtLength: Int,
     maxChunkSize: Int,
     maxChunkCount: Int,
+    maxPartCount: Int,
     maxCommentParsingDepth: Int,
     uriParsingMode: Uri.ParsingMode,
     cookieParsingMode: CookieParsingMode,
@@ -73,6 +74,7 @@ private[pekko] final case class ParserSettingsImpl(
   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(maxPartCount > 0, "max-part-count must be > 0")
   require(maxCommentParsingDepth > 0, "max-comment-parsing-depth must be > 0")
 
   override val defaultHeaderValueCacheLimit: Int = headerValueCacheLimits("default")
@@ -115,6 +117,7 @@ object ParserSettingsImpl extends SettingsCompanionImpl[ParserSettingsImpl]("pek
       c.getIntBytes("max-chunk-ext-length"),
       c.getIntBytes("max-chunk-size"),
       c.getIntBytes("max-chunk-count"),
+      c.getIntBytes("max-part-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 03df29a10..fb94f1cf7 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
@@ -40,6 +40,13 @@ abstract class ParserSettings private[pekko] () extends BodyPartParser.Settings
   def getMaxHeaderNameLength: Int
   def getMaxHeaderValueLength: Int
   def getMaxHeaderCount: Int
+
+  /**
+   * The maximum number of body parts a multipart entity may consist of.
+   *
+   * @since 2.0.0
+   */
+  def getMaxPartCount: Int
   def getMaxContentLength: Long
   def getMaxToStrictBytes: Long
   def getMaxChunkExtLength: Int
@@ -76,6 +83,11 @@ abstract class ParserSettings private[pekko] () extends BodyPartParser.Settings
   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)
+
+  /**
+   * @since 2.0.0
+   */
+  def withMaxPartCount(newValue: Int): ParserSettings = self.copy(maxPartCount = 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 23ee3c75c..07cb9404e 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
@@ -48,6 +48,13 @@ abstract class ParserSettings private[pekko] () extends pekko.http.javadsl.setti
   def maxChunkExtLength: Int
   def maxChunkSize: Int
   def maxChunkCount: Int
+
+  /**
+   * The maximum number of body parts a multipart entity may consist of.
+   *
+   * @since 2.0.0
+   */
+  def maxPartCount: Int
   def maxCommentParsingDepth: Int
   def uriParsingMode: Uri.ParsingMode
   def cookieParsingMode: ParserSettings.CookieParsingMode
@@ -70,6 +77,11 @@ abstract class ParserSettings private[pekko] () extends pekko.http.javadsl.setti
   override def getHeaderValueCacheLimits: util.Map[String, Int] = this.headerValueCacheLimits.asJava
   override def getMaxChunkExtLength = this.maxChunkExtLength
   override def getMaxChunkCount = this.maxChunkCount
+
+  /**
+   * @since 2.0.0
+   */
+  override def getMaxPartCount = this.maxPartCount
   override def getUriParsingMode: pekko.http.javadsl.model.Uri.ParsingMode = this.uriParsingMode
   override def getMaxHeaderCount = this.maxHeaderCount
   override def getMaxContentLength = this.maxContentLength
@@ -114,6 +126,11 @@ abstract class ParserSettings private[pekko] () extends pekko.http.javadsl.setti
   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)
+
+  /**
+   * @since 2.0.0
+   */
+  override def withMaxPartCount(newValue: Int): ParserSettings = self.copy(maxPartCount = 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-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/MultipartUnmarshallersSpec.scala b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/MultipartUnmarshallersSpec.scala
index d5cf37c32..01e6c63ae 100644
--- a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/MultipartUnmarshallersSpec.scala
+++ b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/MultipartUnmarshallersSpec.scala
@@ -19,6 +19,7 @@ import scala.concurrent.duration._
 import org.apache.pekko
 import pekko.http.impl.util._
 import pekko.http.scaladsl.model._
+import pekko.http.scaladsl.settings.ParserSettings
 import pekko.http.scaladsl.model.MediaTypes._
 import pekko.http.scaladsl.model.headers._
 import pekko.http.scaladsl.util.FastFuture._
@@ -253,6 +254,30 @@ trait MultipartUnmarshallersSpec extends PekkoSpecWithMaterializer {
                        |just preamble text""".stripMarginWithNewline(lineFeed))))
             .to[Multipart.General].failed, 1.second.dilated).getMessage shouldEqual "Unexpected end of multipart entity"
       }
+      "more parts than the configured limit" in {
+        implicit val parserSettings: ParserSettings = ParserSettings(system).withMaxPartCount(2)
+        val singlePart =
+          """--12345
+            |
+            |data
+            |""".stripMarginWithNewline(lineFeed)
+
+        Await.result(
+          Unmarshal(HttpEntity(`multipart/mixed`.withBoundary("12345"), ByteString(singlePart * 3 + "--12345--")))
+            .to[Multipart.General].failed,
+          1.second.dilated).getMessage shouldEqual
+        "multipart entity contains more than the configured limit of 2 parts"
+      }
+      "more empty parts than the configured limit" in {
+        implicit val parserSettings: ParserSettings = ParserSettings(system).withMaxPartCount(2)
+        val body = ("--12345" + lineFeed) * 5 + "--12345--"
+
+        Await.result(
+          Unmarshal(HttpEntity(`multipart/mixed`.withBoundary("12345"), ByteString(body)))
+            .to[Multipart.General].failed,
+          1.second.dilated).getMessage shouldEqual
+        "multipart entity contains more than the configured limit of 2 parts"
+      }
       "a stray boundary" in {
         Await.result(
           Unmarshal(HttpEntity(