diff --git a/akka-http-core/src/main/mima-filters/10.1.4.backwards.excludes b/akka-http-core/src/main/mima-filters/10.1.4.backwards.excludes
new file mode 100644
index 00000000000..fd8f3e4a935
--- /dev/null
+++ b/akka-http-core/src/main/mima-filters/10.1.4.backwards.excludes
@@ -0,0 +1,3 @@
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.http.scaladsl.settings.RoutingSettings.decodeMaxSize")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.http.impl.settings.RoutingSettingsImpl.*")
+ProblemFilters.exclude[IncompatibleResultTypeProblem]("akka.http.impl.settings.RoutingSettingsImpl.*")
diff --git a/akka-http-core/src/main/scala/akka/http/impl/settings/RoutingSettingsImpl.scala b/akka-http-core/src/main/scala/akka/http/impl/settings/RoutingSettingsImpl.scala
index 46c194984b4..00f080e4321 100644
--- a/akka-http-core/src/main/scala/akka/http/impl/settings/RoutingSettingsImpl.scala
+++ b/akka-http-core/src/main/scala/akka/http/impl/settings/RoutingSettingsImpl.scala
@@ -17,6 +17,7 @@ private[http] final case class RoutingSettingsImpl(
   rangeCountLimit:          Int,
   rangeCoalescingThreshold: Long,
   decodeMaxBytesPerChunk:   Int,
+  decodeMaxSize:            Long,
   fileIODispatcher:         String) extends akka.http.scaladsl.settings.RoutingSettings {
 
   override def productPrefix = "RoutingSettings"
@@ -30,5 +31,6 @@ object RoutingSettingsImpl extends SettingsCompanion[RoutingSettingsImpl]("akka.
     c getInt "range-count-limit",
     c getBytes "range-coalescing-threshold",
     c getIntBytes "decode-max-bytes-per-chunk",
+    c getPossiblyInfiniteBytes "decode-max-size",
     c getString "file-io-dispatcher")
 }
diff --git a/akka-http-core/src/main/scala/akka/http/javadsl/settings/RoutingSettings.scala b/akka-http-core/src/main/scala/akka/http/javadsl/settings/RoutingSettings.scala
index d6b0fcdff8c..9b4d5bdf00e 100644
--- a/akka-http-core/src/main/scala/akka/http/javadsl/settings/RoutingSettings.scala
+++ b/akka-http-core/src/main/scala/akka/http/javadsl/settings/RoutingSettings.scala
@@ -28,6 +28,7 @@ abstract class RoutingSettings private[akka] () { self: RoutingSettingsImpl ⇒
   def withRangeCountLimit(rangeCountLimit: Int): RoutingSettings = self.copy(rangeCountLimit = rangeCountLimit)
   def withRangeCoalescingThreshold(rangeCoalescingThreshold: Long): RoutingSettings = self.copy(rangeCoalescingThreshold = rangeCoalescingThreshold)
   def withDecodeMaxBytesPerChunk(decodeMaxBytesPerChunk: Int): RoutingSettings = self.copy(decodeMaxBytesPerChunk = decodeMaxBytesPerChunk)
+  def withDecodeMaxSize(decodeMaxSize: Long): RoutingSettings = self.copy(decodeMaxSize = decodeMaxSize)
   def withFileIODispatcher(fileIODispatcher: String): RoutingSettings = self.copy(fileIODispatcher = fileIODispatcher)
 }
 
diff --git a/akka-http-core/src/main/scala/akka/http/scaladsl/settings/RoutingSettings.scala b/akka-http-core/src/main/scala/akka/http/scaladsl/settings/RoutingSettings.scala
index 0a9c217e4b9..19825a688b4 100644
--- a/akka-http-core/src/main/scala/akka/http/scaladsl/settings/RoutingSettings.scala
+++ b/akka-http-core/src/main/scala/akka/http/scaladsl/settings/RoutingSettings.scala
@@ -19,6 +19,7 @@ abstract class RoutingSettings private[akka] () extends akka.http.javadsl.settin
   def rangeCountLimit: Int
   def rangeCoalescingThreshold: Long
   def decodeMaxBytesPerChunk: Int
+  def decodeMaxSize: Long
   def fileIODispatcher: String
 
   /* Java APIs */
@@ -28,6 +29,7 @@ abstract class RoutingSettings private[akka] () extends akka.http.javadsl.settin
   def getRangeCountLimit: Int = rangeCountLimit
   def getRangeCoalescingThreshold: Long = rangeCoalescingThreshold
   def getDecodeMaxBytesPerChunk: Int = decodeMaxBytesPerChunk
+  def getDecodeMaxSize: Long = decodeMaxSize
   def getFileIODispatcher: String = fileIODispatcher
 
   override def withVerboseErrorMessages(verboseErrorMessages: Boolean): RoutingSettings = self.copy(verboseErrorMessages = verboseErrorMessages)
@@ -36,6 +38,7 @@ abstract class RoutingSettings private[akka] () extends akka.http.javadsl.settin
   override def withRangeCountLimit(rangeCountLimit: Int): RoutingSettings = self.copy(rangeCountLimit = rangeCountLimit)
   override def withRangeCoalescingThreshold(rangeCoalescingThreshold: Long): RoutingSettings = self.copy(rangeCoalescingThreshold = rangeCoalescingThreshold)
   override def withDecodeMaxBytesPerChunk(decodeMaxBytesPerChunk: Int): RoutingSettings = self.copy(decodeMaxBytesPerChunk = decodeMaxBytesPerChunk)
+  override def withDecodeMaxSize(decodeMaxSize: Long): RoutingSettings = self.copy(decodeMaxSize = decodeMaxSize)
   override def withFileIODispatcher(fileIODispatcher: String): RoutingSettings = self.copy(fileIODispatcher = fileIODispatcher)
 }
 
diff --git a/akka-http-core/src/test/java/akka/http/javadsl/settings/RoutingSettingsTest.java b/akka-http-core/src/test/java/akka/http/javadsl/settings/RoutingSettingsTest.java
index b6531887c4d..be12e795bfa 100644
--- a/akka-http-core/src/test/java/akka/http/javadsl/settings/RoutingSettingsTest.java
+++ b/akka-http-core/src/test/java/akka/http/javadsl/settings/RoutingSettingsTest.java
@@ -22,6 +22,7 @@ public void testCreateWithActorSystem() {
             "  range-coalescing-threshold = 80\n" +
             "  range-count-limit = 16\n" +
             "  decode-max-bytes-per-chunk = 1m\n" +
+            "  decode-max-size = 8m\n" +
             "  file-io-dispatcher = \"test-only\"\n" +
             "}";
         Config config = ConfigFactory.parseString(testConfig);
diff --git a/akka-http-core/src/test/scala/akka/http/impl/engine/parsing/RequestParserSpec.scala b/akka-http-core/src/test/scala/akka/http/impl/engine/parsing/RequestParserSpec.scala
index d0f5f6f972b..c0e6b052986 100644
--- a/akka-http-core/src/test/scala/akka/http/impl/engine/parsing/RequestParserSpec.scala
+++ b/akka-http-core/src/test/scala/akka/http/impl/engine/parsing/RequestParserSpec.scala
@@ -38,7 +38,7 @@ abstract class RequestParserSpec(mode: String, newLine: String) extends FreeSpec
     akka.loglevel = WARNING
     akka.http.parsing.max-header-value-length = 32
     akka.http.parsing.max-uri-length = 40
-    akka.http.parsing.max-content-length = 4000000000""")
+    akka.http.parsing.max-content-length = infinite""")
   implicit val system = ActorSystem(getClass.getSimpleName, testConf)
   import system.dispatcher
 
diff --git a/akka-http-core/src/test/scala/akka/http/scaladsl/settings/SettingsEqualitySpec.scala b/akka-http-core/src/test/scala/akka/http/scaladsl/settings/SettingsEqualitySpec.scala
index f5dfb3c340f..7ab0485c6a2 100644
--- a/akka-http-core/src/test/scala/akka/http/scaladsl/settings/SettingsEqualitySpec.scala
+++ b/akka-http-core/src/test/scala/akka/http/scaladsl/settings/SettingsEqualitySpec.scala
@@ -19,6 +19,7 @@ class SettingsEqualitySpec extends WordSpec with Matchers {
       range-coalescing-threshold = 80
       range-count-limit = 16
       decode-max-bytes-per-chunk = 1m
+      decode-max-size = 8m
       file-io-dispatcher = ${akka.stream.blocking-io-dispatcher}
     }
   """).withFallback(ConfigFactory.load).resolve
diff --git a/akka-http-tests/src/test/scala/akka/http/scaladsl/server/SizeLimitSpec.scala b/akka-http-tests/src/test/scala/akka/http/scaladsl/server/SizeLimitSpec.scala
new file mode 100644
index 00000000000..f603b88d61e
--- /dev/null
+++ b/akka-http-tests/src/test/scala/akka/http/scaladsl/server/SizeLimitSpec.scala
@@ -0,0 +1,241 @@
+/*
+ * Copyright (C) 2009-2018 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.http.scaladsl.server
+
+import akka.NotUsed
+
+import scala.collection.immutable
+import akka.actor.ActorSystem
+import akka.http.scaladsl.client.RequestBuilding
+import akka.http.scaladsl.coding.{ Decoder, Gzip }
+import akka.http.scaladsl.model._
+import akka.http.scaladsl.server.Directives._
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.HttpEntity.Chunk
+import akka.http.scaladsl.model.headers.{ HttpEncoding, HttpEncodings, `Content-Encoding` }
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.{ Flow, Source }
+import akka.testkit.{ EventFilter, TestKit }
+import akka.util.ByteString
+import com.typesafe.config.{ Config, ConfigFactory }
+import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpec }
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.time.{ Millis, Seconds, Span }
+
+class SizeLimitSpec extends WordSpec with Matchers with RequestBuilding with BeforeAndAfterAll with ScalaFutures {
+
+  val maxContentLength = 800
+  // Protect network more than memory:
+  val decodeMaxSize = 1600
+
+  val testConf: Config = ConfigFactory.parseString(s"""
+    akka.loggers = ["akka.testkit.TestEventListener"]
+    akka.loglevel = ERROR
+    akka.stdout-loglevel = ERROR
+    akka.http.parsing.max-content-length = $maxContentLength
+    akka.http.routing.decode-max-size = $decodeMaxSize
+    """)
+  implicit val system = ActorSystem(getClass.getSimpleName, testConf)
+  import system.dispatcher
+  implicit val materializer = ActorMaterializer()
+  val random = new scala.util.Random(42)
+
+  implicit val defaultPatience = PatienceConfig(timeout = Span(2, Seconds), interval = Span(5, Millis))
+
+  "a normal route" should {
+    val route = path("noDirective") {
+      post {
+        entity(as[String]) { _ ⇒
+          complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, "<h1>Say hello to akka-http</h1>"))
+        }
+      }
+    }
+
+    val binding = Http().bindAndHandle(route, "localhost", port = 0).futureValue
+
+    "accept small POST requests" in {
+      Http().singleRequest(Post(s"http:/${binding.localAddress}/noDirective", entityOfSize(maxContentLength)))
+        .futureValue.status shouldEqual StatusCodes.OK
+    }
+
+    "not accept entities bigger than configured with akka.http.parsing.max-content-length" in {
+      Http().singleRequest(Post(s"http:/${binding.localAddress}/noDirective", entityOfSize(maxContentLength + 1)))
+        .futureValue.status shouldEqual StatusCodes.BadRequest
+    }
+  }
+
+  "a route with decodeRequest" should {
+    val route = path("noDirective") {
+      decodeRequest {
+        post {
+          entity(as[String]) { e ⇒
+            complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, s"Got request with entity of ${e.length} characters"))
+          }
+        }
+      }
+    }
+
+    val binding = Http().bindAndHandle(route, "localhost", port = 0).futureValue
+
+    "accept a small request" in {
+      val response = Http().singleRequest(Post(s"http:/${binding.localAddress}/noDirective", entityOfSize(maxContentLength))).futureValue
+      response.status shouldEqual StatusCodes.OK
+      response.entity.dataBytes.runReduce(_ ++ _).futureValue.utf8String shouldEqual (s"Got request with entity of $maxContentLength characters")
+    }
+
+    "reject a small request that decodes into a large entity" in {
+      val data = ByteString.fromString("0" * (decodeMaxSize + 1))
+      val zippedData = Gzip.encode(data)
+      val request = HttpRequest(
+        HttpMethods.POST,
+        s"http:/${binding.localAddress}/noDirective",
+        immutable.Seq(`Content-Encoding`(HttpEncodings.gzip)),
+        HttpEntity(ContentTypes.`text/plain(UTF-8)`, zippedData))
+
+      zippedData.size should be <= maxContentLength
+      data.size should be > decodeMaxSize
+
+      Http().singleRequest(request)
+        .futureValue.status shouldEqual StatusCodes.BadRequest
+    }
+  }
+
+  "a route with decodeRequest that results in a large chunked entity" should {
+    val decoder = decodeTo(chunkedEntityOfSize(decodeMaxSize + 1))
+
+    val route = path("noDirective") {
+      decodeRequestWith(decoder) {
+        post {
+          entity(as[String]) { e ⇒
+            complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, s"Got request with entity of ${e.length} characters"))
+          }
+        }
+      }
+    }
+
+    val binding = Http().bindAndHandle(route, "localhost", port = 0).futureValue
+
+    "reject a small request that decodes into a large chunked entity" in {
+      val request = Post(s"http:/${binding.localAddress}/noDirective", "x").withHeaders(`Content-Encoding`(HttpEncoding("custom")))
+      val response = Http().singleRequest(request).futureValue
+      response.status shouldEqual StatusCodes.BadRequest
+    }
+  }
+
+  "a route with decodeRequest that results in a large non-chunked streaming entity" should {
+    val decoder = decodeTo(nonChunkedEntityOfSize(decodeMaxSize + 1))
+
+    val route = path("noDirective") {
+      decodeRequestWith(decoder) {
+        post {
+          entity(as[String]) { e ⇒
+            complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, s"Got request with entity of ${e.length} characters"))
+          }
+        }
+      }
+    }
+
+    val binding = Http().bindAndHandle(route, "localhost", port = 0).futureValue
+
+    "reject a small request that decodes into a large non-chunked streaming entity" in {
+      val request = Post(s"http:/${binding.localAddress}/noDirective", "x").withHeaders(`Content-Encoding`(HttpEncoding("custom")))
+      val response = Http().singleRequest(request).futureValue
+      response.status shouldEqual StatusCodes.BadRequest
+    }
+  }
+
+  "a route with decodeRequest followed by withoutSizeLimit" should {
+    val route = path("noDirective") {
+      decodeRequest {
+        withoutSizeLimit {
+          post {
+            entity(as[String]) { e ⇒
+              complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, s"Got request with entity of ${e.length} characters"))
+            }
+          }
+        }
+      }
+    }
+
+    val binding = Http().bindAndHandle(route, "localhost", port = 0).futureValue
+
+    "accept a small request" in {
+      Http().singleRequest(Post(s"http:/${binding.localAddress}/noDirective", entityOfSize(maxContentLength)))
+        .futureValue.status shouldEqual StatusCodes.OK
+    }
+
+    "accept a small request that decodes into a large entity" in {
+      val data = ByteString.fromString("0" * (decodeMaxSize + 1))
+      val zippedData = Gzip.encode(data)
+      val request = HttpRequest(
+        HttpMethods.POST,
+        s"http:/${binding.localAddress}/noDirective",
+        immutable.Seq(`Content-Encoding`(HttpEncodings.gzip)),
+        HttpEntity(ContentTypes.`text/plain(UTF-8)`, zippedData))
+
+      zippedData.size should be <= maxContentLength
+      data.size should be > decodeMaxSize
+
+      val response = Http().singleRequest(request).futureValue
+      response.status shouldEqual StatusCodes.OK
+      response.entity.dataBytes.runReduce(_ ++ _).futureValue.utf8String shouldEqual (s"Got request with entity of ${decodeMaxSize + 1} characters")
+    }
+
+    "accept a large request that decodes into a large entity" in {
+      val data = new Array[Byte](decodeMaxSize)
+      random.nextBytes(data)
+      val zippedData = Gzip.encode(ByteString(data))
+      val request = HttpRequest(
+        HttpMethods.POST,
+        s"http:/${binding.localAddress}/noDirective",
+        immutable.Seq(`Content-Encoding`(HttpEncodings.gzip)),
+        HttpEntity(ContentTypes.`text/plain(UTF-8)`, zippedData))
+
+      zippedData.size should be > maxContentLength
+      data.length should be <= decodeMaxSize
+
+      Http().singleRequest(request)
+        .futureValue.status shouldEqual StatusCodes.OK
+    }
+  }
+
+  "the withoutSizeLimit directive" should {
+    val route = path("withoutSizeLimit") {
+      post {
+        withoutSizeLimit {
+          entity(as[String]) { _ ⇒
+            complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, "<h1>Say hello to akka-http</h1>"))
+          }
+        }
+      }
+    }
+
+    val binding = Http().bindAndHandle(route, "localhost", port = 0).futureValue
+
+    "accept entities bigger than configured with akka.http.parsing.max-content-length" in {
+      Http().singleRequest(Post(s"http:/${binding.localAddress}/withoutSizeLimit", entityOfSize(maxContentLength + 1)))
+        .futureValue.status shouldEqual StatusCodes.OK
+    }
+  }
+
+  override def afterAll() = TestKit.shutdownActorSystem(system)
+
+  private def byteSource(size: Int): Source[ByteString, Any] = Source(Array.fill[ByteString](size)(ByteString("0")).toVector)
+
+  private def chunkedEntityOfSize(size: Int) = HttpEntity.Chunked(ContentTypes.`text/plain(UTF-8)`, byteSource(size).map(Chunk(_)))
+  private def nonChunkedEntityOfSize(size: Int): MessageEntity = HttpEntity.Default(ContentTypes.`text/plain(UTF-8)`, size, byteSource(size))
+  private def entityOfSize(size: Int) = HttpEntity(ContentTypes.`text/plain(UTF-8)`, "0" * size)
+
+  private def decodeTo(result: MessageEntity): Decoder = new Decoder {
+    override def encoding: HttpEncoding = HttpEncoding("custom")
+
+    override def maxBytesPerChunk: Int = 1000
+    override def withMaxBytesPerChunk(maxBytesPerChunk: Int): Decoder = this
+
+    override def decoderFlow: Flow[ByteString, ByteString, NotUsed] = ???
+
+    override def decodeMessage(message: HttpMessage) = message.withEntity(result)
+  }
+}
diff --git a/akka-http-tests/src/test/scala/akka/http/scaladsl/server/WithoutSizeLimitSpec.scala b/akka-http-tests/src/test/scala/akka/http/scaladsl/server/WithoutSizeLimitSpec.scala
deleted file mode 100644
index 5a096ae98dc..00000000000
--- a/akka-http-tests/src/test/scala/akka/http/scaladsl/server/WithoutSizeLimitSpec.scala
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * Copyright (C) 2009-2018 Lightbend Inc. <https://www.lightbend.com>
- */
-
-package akka.http.scaladsl.server
-
-import akka.actor.ActorSystem
-import akka.http.scaladsl.client.RequestBuilding
-import akka.http.scaladsl.model._
-import akka.http.scaladsl.server.Directives._
-import akka.http.scaladsl.Http
-import akka.stream.ActorMaterializer
-import akka.testkit.{ SocketUtil, TestKit }
-import com.typesafe.config.{ Config, ConfigFactory }
-import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpec }
-
-import scala.concurrent.Await
-import scala.concurrent.duration._
-
-class WithoutSizeLimitSpec extends WordSpec with Matchers with RequestBuilding with BeforeAndAfterAll {
-  val testConf: Config = ConfigFactory.parseString("""
-    akka.loggers = ["akka.testkit.TestEventListener"]
-    akka.loglevel = ERROR
-    akka.stdout-loglevel = ERROR
-    akka.http.parsing.max-content-length = 800""")
-  implicit val system = ActorSystem(getClass.getSimpleName, testConf)
-  import system.dispatcher
-  implicit val materializer = ActorMaterializer()
-
-  "the withoutSizeLimit directive" should {
-    "accept entities bigger than configured with akka.http.parsing.max-content-length" in {
-      val route =
-        path("noDirective") {
-          post {
-            entity(as[String]) { _ ⇒
-              complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, "<h1>Say hello to akka-http</h1>"))
-            }
-          }
-        } ~
-          path("withoutSizeLimit") {
-            post {
-              withoutSizeLimit {
-                entity(as[String]) { _ ⇒
-                  complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, "<h1>Say hello to akka-http</h1>"))
-                }
-              }
-            }
-          }
-
-      val (hostName, port) = SocketUtil.temporaryServerHostnameAndPort()
-
-      // It went from 1 occurrence to 2 after discarding the entity, I think is due to the retrying nature of `handleRejections`
-      // that causes one Exception for the original entity and another one from the rejected one.
-      val future = for {
-        _ ← Http().bindAndHandle(route, hostName, port)
-
-        requestToNoDirective = Post(s"http://$hostName:$port/noDirective", entityOfSize(801))
-        responseWithoutDirective ← Http().singleRequest(requestToNoDirective)
-        _ = responseWithoutDirective.status shouldEqual StatusCodes.BadRequest
-
-        requestToDirective = Post(s"http://$hostName:$port/withoutSizeLimit", entityOfSize(801))
-        responseWithDirective ← Http().singleRequest(requestToDirective)
-      } yield responseWithDirective
-
-      val response = Await.result(future, 5 seconds)
-      response.status shouldEqual StatusCodes.OK
-    }
-  }
-
-  override def afterAll() = TestKit.shutdownActorSystem(system)
-
-  private def entityOfSize(size: Int) = HttpEntity(ContentTypes.`text/plain(UTF-8)`, "0" * size)
-}
diff --git a/akka-http/src/main/resources/reference.conf b/akka-http/src/main/resources/reference.conf
index 654c1dda9e1..e92628acbba 100644
--- a/akka-http/src/main/resources/reference.conf
+++ b/akka-http/src/main/resources/reference.conf
@@ -37,6 +37,15 @@ akka.http.routing {
   # for an entity data stream.
   decode-max-bytes-per-chunk = 1m
 
+  # Maximum content length after applying a decoding directive. When the directive
+  # decompresses, for example, an entity compressed with gzip, the resulting stream can be much
+  # larger than the max-content-length. Like with max-content-length, this is not necessarilly a
+  # problem when consuming the entity in a streaming fashion, but does risk high memory use
+  # when the entity is made strict or marshalled into an in-memory object.
+  # This limit (like max-content-length) can be overridden on a case-by-case basis using the
+  # withSizeLimit directive.
+  decode-max-size = 8m
+
   # Fully qualified config path which holds the dispatcher configuration
   # to be used by FlowMaterialiser when creating Actors for IO operations.
   file-io-dispatcher = ${akka.stream.blocking-io-dispatcher}
diff --git a/akka-http/src/main/scala/akka/http/scaladsl/coding/Decoder.scala b/akka-http/src/main/scala/akka/http/scaladsl/coding/Decoder.scala
index 639df9cebab..0fa6ed5a76f 100644
--- a/akka-http/src/main/scala/akka/http/scaladsl/coding/Decoder.scala
+++ b/akka-http/src/main/scala/akka/http/scaladsl/coding/Decoder.scala
@@ -13,13 +13,19 @@ import headers.HttpEncoding
 import akka.stream.scaladsl.{ Flow, Sink, Source }
 
 import scala.concurrent.Future
+import scala.util.control.NonFatal
 
 trait Decoder {
   def encoding: HttpEncoding
 
   def decodeMessage(message: HttpMessage): message.Self =
     if (message.headers exists Encoder.isContentEncodingHeader)
-      message.transformEntityDataBytes(decoderFlow).withHeaders(message.headers filterNot Encoder.isContentEncodingHeader)
+      message.transformEntityDataBytes(decoderFlow.recover {
+        case NonFatal(e) ⇒
+          throw IllegalRequestException(
+            StatusCodes.BadRequest,
+            ErrorInfo("The request's encoding is corrupt", e.getMessage))
+      }).withHeaders(message.headers filterNot Encoder.isContentEncodingHeader)
     else message.self
 
   def decodeData[T](t: T)(implicit mapper: DataMapper[T]): T = mapper.transformDataBytes(t, decoderFlow)
diff --git a/akka-http/src/main/scala/akka/http/scaladsl/server/directives/CodingDirectives.scala b/akka-http/src/main/scala/akka/http/scaladsl/server/directives/CodingDirectives.scala
index 5b8bdaf2e12..9e7a3643e69 100644
--- a/akka-http/src/main/scala/akka/http/scaladsl/server/directives/CodingDirectives.scala
+++ b/akka-http/src/main/scala/akka/http/scaladsl/server/directives/CodingDirectives.scala
@@ -80,16 +80,7 @@ trait CodingDirectives {
       else
         extractSettings flatMap { settings ⇒
           val effectiveDecoder = decoder.withMaxBytesPerChunk(settings.decodeMaxBytesPerChunk)
-          mapRequest { request ⇒
-            effectiveDecoder.decodeMessage(request).mapEntity { entity ⇒
-              entity.transformDataBytes(Flow[ByteString].recover {
-                case NonFatal(e) ⇒
-                  throw IllegalRequestException(
-                    StatusCodes.BadRequest,
-                    ErrorInfo("The request's encoding is corrupt", e.getMessage))
-              })
-            }
-          }
+          mapRequest(effectiveDecoder.decodeMessage(_)) & withSizeLimit(settings.decodeMaxSize)
         }
 
     requestEntityEmpty | (
diff --git a/docs/src/main/paradox/routing-dsl/directives/coding-directives/decodeRequest.md b/docs/src/main/paradox/routing-dsl/directives/coding-directives/decodeRequest.md
index e4e60c905b8..6f72b67dd0d 100644
--- a/docs/src/main/paradox/routing-dsl/directives/coding-directives/decodeRequest.md
+++ b/docs/src/main/paradox/routing-dsl/directives/coding-directives/decodeRequest.md
@@ -10,7 +10,11 @@
 
 ## Description
 
-Decompresses the incoming request if it is `gzip` or `deflate` compressed. Uncompressed requests are passed through untouched. If the request encoded with another encoding the request is rejected with an @unidoc[UnsupportedRequestEncodingRejection].
+Decompresses the incoming request if it is `gzip` or `deflate` compressed. Uncompressed requests are passed through untouched.
+If the request encoded with another encoding the request is rejected with an @unidoc[UnsupportedRequestEncodingRejection].
+If the request entity after decoding exceeds `akka.http.routing.decode-max-size` the stream fails with an
+@unidoc[akka.http.scaladsl.model.EntityStreamSizeException].
+
 
 ## Example
 
diff --git a/docs/src/main/paradox/routing-dsl/directives/coding-directives/decodeRequestWith.md b/docs/src/main/paradox/routing-dsl/directives/coding-directives/decodeRequestWith.md
index fe14f097de9..d856b378e7c 100644
--- a/docs/src/main/paradox/routing-dsl/directives/coding-directives/decodeRequestWith.md
+++ b/docs/src/main/paradox/routing-dsl/directives/coding-directives/decodeRequestWith.md
@@ -10,7 +10,11 @@
 
 ## Description
 
-Decodes the incoming request if it is encoded with one of the given encoders. If the request encoding doesn't match one of the given encoders the request is rejected with an @unidoc[UnsupportedRequestEncodingRejection]. If no decoders are given the default encoders (`Gzip`, `Deflate`, `NoCoding`) are used.
+Decodes the incoming request if it is encoded with one of the given encoders.
+If the request encoding doesn't match one of the given encoders the request is rejected with an @unidoc[UnsupportedRequestEncodingRejection]. If no decoders are given the default encoders (`Gzip`, `Deflate`, `NoCoding`) are used.
+If the request entity after decoding exceeds `akka.http.routing.decode-max-size` the stream fails with an
+@unidoc[akka.http.scaladsl.model.EntityStreamSizeException].
+
 
 ## Example