add ability to disable throttle

http-core/src/main/resources/reference.conf

@@ -311,6 +311,7 @@ pekko.http {
       # Configure the throttle for Reset Frames (https://github.com/apache/incubator-pekko-http/issues/332)
       resets-throttle-cost = 100
       resets-throttle-burst = 100
+      # setting resets-throttle-interval to 0s will disable the throttle
       resets-throttle-interval = 1s
     }
 

http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala

@@ -122,12 +122,17 @@ private[http] object Http2Blueprint {
       telemetry: TelemetrySpi,
     dateHeaderRendering: DateHeaderRendering): BidiFlow[HttpResponse, ByteString, ByteString, HttpRequest, ServerTerminator] = {
     val masterHttpHeaderParser = HttpHeaderParser(settings.parserSettings, log) // FIXME: reuse for framing
-    telemetry.serverConnection atop
+    val flow0 = telemetry.serverConnection atop
       httpLayer(settings, log, dateHeaderRendering) atopKeepRight
       serverDemux(settings.http2Settings, initialDemuxerSettings, upgraded) atop
       FrameLogger.logFramesIfEnabled(settings.http2Settings.logFrames) atop // enable for debugging
-      hpackCoding(masterHttpHeaderParser, settings.parserSettings) atop
-      rapidResetMitigation(settings.http2Settings) atop
+      hpackCoding(masterHttpHeaderParser, settings.parserSettings)
+
+    val flow1 = if (settings.http2Settings.resetsThrottleInterval.toMillis > 0) {
+      flow0 atop rapidResetMitigation(settings.http2Settings)
+    } else flow0
+
+    flow1 atop
       framing(log) atop
       errorHandling(log) atop
       idleTimeoutIfConfigured(settings.idleTimeout)

http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ServerDisableResetSpec.scala

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2018-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.http.impl.engine.http2
+
+import org.apache.pekko
+import org.apache.pekko.http.impl.engine.http2.Http2Protocol.FrameType
+import org.apache.pekko.http.impl.engine.http2.framing.FrameRenderer
+import org.apache.pekko.util.ByteStringBuilder
+
+import java.nio.ByteOrder
+
+/**
+ * This tests the http2 server support for rapid resets can be disabled.
+ */
+class Http2ServerDisableResetSpec extends Http2SpecWithMaterializer("""
+    pekko.http.server.remote-address-header = on
+    pekko.http.server.http2.log-frames = on
+    pekko.http.server.http2.resets-throttle-interval = 0s
+  """) {
+  override def failOnSevereMessages: Boolean = true
+
+  "The Http/2 server implementation" should {
+    "not cancel connection during rapid reset attack (throttle disabled)".inAssertAllStagesStopped(new TestSetup with RequestResponseProbes {
+      implicit val bigEndian: ByteOrder = ByteOrder.BIG_ENDIAN
+      val bb = new ByteStringBuilder
+      bb.putInt(0)
+      val rstFrame = FrameRenderer.renderFrame(FrameType.RST_STREAM, ByteFlag.Zero, 1, bb.result())
+      val longFrame = Seq.fill(1000)(rstFrame).reduce(_ ++ _)
+      network.sendBytes(longFrame)
+    })
+  }
+}